This article is a practical guide for developers on creating autonomous AI agents in Python. We will not repeat the theory of what LangChain and LangGraph are. Instead, we will focus on code, architecture, and solving real-world problems.
Goal: To build two projects from scratch:
- A Classifier Agent: A multi-step agent with managed state, but without external tools.
- An Assistant Agent: A full-fledged agent with access to the file system and web search via the MCP protocol, built on cyclical logic.
We will cover best practices: configuration management, model selection, and error handling to create robust systems.
A Brief on the Concepts: The Agent and the MCP Bridge
Before diving into the code, let’s establish two concepts:
- AI Agent: A program built around a “reason-action” loop. It receives a task, uses an LLM to decide what to do next (e.g., call a tool), executes the action, and repeats the cycle until the task is complete.
- MCP (Model Context Protocol): A standard that acts as a bridge between the agent’s logic and external tools. It allows the agent to work with files, APIs, or search in a unified way, without worrying about their implementation details.
Part 1: Setting up a Robust Environment
Step 1: Virtual Environment and Dependencies
Create and activate a virtual environment. Then, create a requirements.txt file:
# Core Frameworks
langchain
langgraph
# Model Adapters
langchain-openai
langchain-google-genai
langchain-mistralai
langchain-community # For Ollama
# Tools and Protocols
langchain-mcp-adapters
mcp
ollama
# Utilities
python-dotenv
tenacity # For robust error handlingInstall the dependencies:
pip install -r requirements.txt```
#### Step 2: API Key Configuration
Create a `.env` file to store your keys:OPENAI_API_KEY=”sk-…”
GOOGLE_API_KEY=”AIzaSy…”
MISTRAL_API_KEY=”…”
BRAVE_API_KEY=”…” # For the web search tool via MCP
#### Step 3: The "Model Factory" Pattern
To flexibly switch between cloud and local models without changing the agent's code, we'll use the factory pattern.python
llm_factory.py
import os
from enum import Enum
from dotenv import load_dotenv
from langchain_openai import ChatOpenAI
from langchain_google_genai import ChatGoogleGenerativeAI
from langchain_mistralai import ChatMistralAI
from langchain_community.chat_models import ChatOllama
load_dotenv()
class ModelProvider(Enum):
OPENAI = “openai”
GEMINI = “gemini”
MISTRAL_API = “mistral_api”
OLLAMA = “ollama”
def get_llm(provider: ModelProvider, model_name: str = None):
“””A factory for creating LLM instances.”””
if provider == ModelProvider.OPENAI:
return ChatOpenAI(model=model_name or “gpt-4o-mini”, temperature=0)
elif provider == ModelProvider.GEMINI:
return ChatGoogleGenerativeAI(model=model_name or “gemini-1.5-flash”, temperature=0)
elif provider == ModelProvider.MISTRAL_API:
return ChatMistralAI(model=model_name or “mistral-large-latest”, temperature=0)
elif provider == ModelProvider.OLLAMA:
# Ensure you have Ollama running with the required model
# docker exec -it ollama ollama pull mistral
return ChatOllama(model=model_name or “mistral”, temperature=0)
raise ValueError(f”Unknown model provider: {provider}”)
Example usage
if name == “main“:
# local_llm = get_llm(ModelProvider.OLLAMA)
openai_llm = get_llm(ModelProvider.OPENAI)
response = openai_llm.invoke(“Explain the concept of RAG in three sentences.”)
print(response.content)
### Part 2: Project 1 — Vacancy Classification Agent
This agent demonstrates how to use LangGraph to create a **linear graph** with managed state. It will take a job description and sequentially classify it based on three parameters.
#### Step 1: Defining the State
The state is the "memory" of our graph, passed from one node to the next.python
vacancy_classifier.py
from typing import TypedDict, Dict
class ClassificationState(TypedDict):
“””State for the classifier agent.”””
description: str # Source text
job_type: str # Job type (project/permanent)
category: str # Profession
search_type: str # Goal (looking for a job/performer)
classification_log: list # Debug log
#### Step 2: Implementing the Graph Nodes
Each node is a function that takes the state, performs its part of the work, and returns the updated state.python
import asyncio
import json
from langchain_core.prompts import ChatPromptTemplate
from llm_factory import get_llm, ModelProvider
class VacancyClassifierAgent:
def init(self):
self.llm = get_llm(ModelProvider.OPENAI, model_name=”gpt-4o-mini”)
async def _classify_job_type(self, state: ClassificationState) -> ClassificationState:
"""Node 1: Determines the job type."""
prompt = ChatPromptTemplate.from_messages([
("system", "Determine the job type. The answer must be 'project-based' or 'permanent'."),
("human", "Job description:\n\n{description}")
])
chain = prompt | self.llm
result = await chain.ainvoke({"description": state["description"]})
state["job_type"] = result.content.strip()
state["classification_log"].append("Determined job type.")
return state
async def _classify_category(self, state: ClassificationState) -> ClassificationState:
"""Node 2: Determines the profession category."""
# Categories can be loaded from a file or database
categories = ["Python Developer", "Designer", "Marketer", "3D Animator"]
prompt = ChatPromptTemplate.from_messages([
("system", f"Choose the most suitable category from the list: {', '.join(categories)}."),
("human", "Job description:\n\n{description}")
])
chain = prompt | self.llm
result = await chain.ainvoke({"description": state["description"]})
state["category"] = result.content.strip()
state["classification_log"].append("Determined category.")
return state
async def _classify_search_type(self, state: ClassificationState) -> ClassificationState:
"""Node 3: Determines the search goal."""
prompt = ChatPromptTemplate.from_messages([
("system", "Determine the author's goal. The answer must be 'looking for a job' or 'looking for a performer'."),
("human", "Job description:\n\n{description}")
])
chain = prompt | self.llm
result = await chain.ainvoke({"description": state["description"]})
state["search_type"] = result.content.strip()
state["classification_log"].append("Determined search goal.")
return state#### Step 3: Assembling and Running the Graph
We assemble the nodes into a single workflow.python
… continuation of the VacancyClassifierAgent class …
from langgraph.graph import StateGraph, END
def build_graph(self):
"""Assembles the state graph."""
workflow = StateGraph(ClassificationState)
workflow.add_node("job_type_classifier", self._classify_job_type)
workflow.add_node("category_classifier", self._classify_category)
workflow.add_node("search_type_classifier", self._classify_search_type)
workflow.set_entry_point("job_type_classifier")
workflow.add_edge("job_type_classifier", "category_classifier")
workflow.add_edge("category_classifier", "search_type_classifier")
workflow.add_edge("search_type_classifier", END)
return workflow.compile()async def main():
agent = VacancyClassifierAgent()
graph = agent.build_graph()
description = "We are looking for an experienced Python developer to join our team full-time to work on a fintech project."
initial_state = ClassificationState(
description=description,
job_type="", category="", search_type="",
classification_log=[]
)
final_state = await graph.ainvoke(initial_state)
print("--- Classification Result ---")
print(json.dumps(final_state, indent=2, ensure_ascii=False))if name == “main“:
asyncio.run(main())
### Part 3: Project 2 — Assistant Agent with Tools (MCP)
This agent demonstrates **cyclical logic**, where it can repeatedly call tools to solve a task.
#### Step 1: Configuration Management
For agents interacting with the external world, robust configuration is essential.python
mcp_agent_config.py
from dataclasses import dataclass, field
import os
from llm_factory import ModelProvider
@dataclass
class AgentConfig:
workdir: str = “./agent_workdir”
model_provider: ModelProvider = ModelProvider.OLLAMA
def __post_init__(self):
"""Post-initialization validation."""
os.makedirs(self.workdir, exist_ok=True)#### Step 2: Defining the State for Dialogue
The state will now store the message history.python
mcp_agent.py
from typing import TypedDict, Annotated, Sequence
from langchain_core.messages import BaseMessage
import operator
class AgentState(TypedDict):
messages: Annotated[Sequence[BaseMessage], operator.add]
#### Step 3: Implementing the Cyclical Graph
The graph will consist of two main nodes and a conditional edge that creates the "reason-action" loop.python
from langgraph.graph import StateGraph, END
from langgraph.prebuilt import ToolExecutor
from langchain_mcp_adapters.langchain import V1ToolExecutor
from langchain_mcp_adapters.clients import MultiServerMCPClient
from llm_factory import get_llm
from mcp_agent_config import AgentConfig
class MCPAgent:
def init(self, config: AgentConfig):
self.config = config
self.llm = get_llm(config.model_provider)
self.tools = []
self.tool_executor = None
async def setup_tools(self):
"""Initializes tools via MCP."""
mcp_config = {
"filesystem": {
"command": "npx", "args": ["-y", "@modelcontextprotocol/server-filesystem", self.config.workdir],
"transport": "stdio"
},
# Add brave-search if you have a BRAVE_API_KEY
}
mcp_client = MultiServerMCPClient(mcp_config)
self.tools = await mcp_client.get_tools()
self.tool_executor = ToolExecutor([V1ToolExecutor(tool) for tool in self.tools])
# Bind the tools to the model
self.llm = self.llm.bind_tools(self.tools)
def _should_continue(self, state: AgentState):
"""Conditional edge: decides whether to call a tool."""
last_message = state['messages'][-1]
if not last_message.tool_calls:
return "end"
return "continue"
def _call_model(self, state: AgentState):
"""Node 1: Call the LLM to make a decision."""
response = self.llm.invoke(state['messages'])
return {"messages": [response]}
def _call_tool(self, state: AgentState):
"""Node 2: Execute the tool call."""
last_message = state['messages'][-1]
tool_call = last_message.tool_calls[0]
action = {"tool": tool_call["name"], "tool_input": tool_call["args"], "log": ""}
response = self.tool_executor.invoke(action)
return {"messages": [response]}
def build_graph(self):
workflow = StateGraph(AgentState)
workflow.add_node("agent", self._call_model)
workflow.add_node("action", self._call_tool)
workflow.set_entry_point("agent")
workflow.add_conditional_edges(
"agent",
self._should_continue,
{"continue": "action", "end": END}
)
workflow.add_edge("action", "agent")
return workflow.compile()#### Step 4: Running and Interactingpython
… continuation of mcp_agent.py …
import asyncio
from langchain_core.messages import HumanMessage
from tenacity import retry, stop_after_attempt, wait_fixed
@retry(stop=stop_after_attempt(3), wait=wait_fixed(1))
async def run_agent_task(graph, task):
“””Runs a task with error handling.”””
return await graph.ainvoke({“messages”: [HumanMessage(content=task)]})
async def main():
config = AgentConfig(model_provider=ModelProvider.OPENAI) # or OLLAMA
agent = MCPAgent(config)
await agent.setup_tools()
graph = agent.build_graph()
task = "Create a file named 'hello.txt' in the working directory and write 'Hello, world!' into it."
result = await run_agent_task(graph, task)
print("\n--- Agent's Final Response ---")
print(result['messages'][-1].content)if name == “main“:
asyncio.run(main())`` Here, we've added thetenacity` decorator for robustness—if the agent call fails due to a temporary network error, it will be automatically retried.
Conclusion
We have built two types of agents using modern practices:
- A linear graph is excellent for tasks with a clear sequence of steps, such as ETL processes or multi-stage analysis.
- A cyclical graph is the foundation for creating interactive assistants and autonomous agents capable of solving complex problems with tools.
The architectural patterns presented—the model factory, configuration management, separating logic into nodes, and using state graphs—are the fundamentals for building scalable and robust AI systems.