Module 7: Advanced MCP and Tool Integration
8. Project: Research Agent v4 — MCP Integration
Project Overview
In Module 6 you built Research Agent v3: durable checkpointing with PostgresSaver, long-term memory with InMemoryStore, conversation management with trimming and summarization, and time-travel debugging. It's an agent that plans, researches, reflects, remembers across sessions, and survives restarts. But it has a problem that becomes obvious once you want it to do more: its tools are hardcoded.
If you want the agent to save research to disk, you need to write a filesystem @tool, import it, and re-deploy. If you want it to search academic papers, you write another @tool, import, re-deploy. If you want to switch from Tavily to DuckDuckGo for web search, you rewrite the existing tool. Every new tool requires modifying the agent's code.
In this project you turn v3 into Research Agent v4 by replacing the hardcoded tools with 3 independent MCP servers:
-
Filesystem MCP Server: Reads and writes research files — notes, summaries, sources. The agent can save research to disk and read previous research without the tool living in its code.
-
Web Search MCP Server: A wrapper around DuckDuckGo as an MCP server. It replaces M2's hardcoded
web_search. If tomorrow you want to switch to Tavily or Brave Search, you deploy a new server without touching the agent. -
Paper Database MCP Server: A custom server that exposes search and lookup of academic papers. It shows how to create an MCP server for your own data sources.
-
Dynamic tool discovery: At startup, the agent connects to the 3 servers, discovers every available tool, and uses them inside its graph — without knowing in advance which tools exist.
Estimated time: 60-90 minutes.
Project Goal
Replace Research Agent v3's hardcoded tools with tools discovered dynamically from 3 MCP servers, keeping the planning, reflection, memory, and checkpointing architecture intact.
By the end you'll be able to:
- Create 3 independent MCP servers with FastMCP (filesystem, web search, papers)
- Connect the Research Agent to all 3 servers using
langchain-mcp-adapters - Discover tools dynamically when the agent starts
- Run complete research runs using MCP tools instead of hardcoded tools
- Add a new tool to a server without changing a single line of the agent's code
- Verify that checkpointing, long-term memory, and conversation management still work with MCP tools
What Changes vs v3 (M6)
Research Agent v3's foundation stays intact: planning with priorities, iterative research, analysis, synthesis, reflection with quality gates, re-planning, checkpointing, long-term memory, conversation management. You don't rewrite those nodes. What changes is where the tools come from.
Before (v3): hardcoded tools
from langchain_community.tools.tavily_search import TavilySearchResults
search_tool = TavilySearchResults(max_results=3)
tools = [search_tool]
model = ChatOpenAI(model="gpt-4.1").bind_tools(tools)After (v4): tools discovered via MCP
from langchain_mcp_adapters.client import MultiServerMCPClient
async with MultiServerMCPClient(server_config) as mcp_client:
tools = mcp_client.get_tools()
# 8 tools discovered from 3 servers — no imports, no hardcoding
model = ChatOpenAI(model="gpt-4.1").bind_tools(tools)New state fields
| Field | Type | What for |
|---|---|---|
available_tools | list[str] | The list of tools discovered at startup |
active_mcp_servers | list[str] | The servers that connected successfully |
tool_source | dict[str, str] | A tool_name → server_name map |
Changes to the flow
v3: [summarize] → planning → research ⇄ analysis → synthesis → reflection
↑ │
trim + resume replan ◄──┘
+ PostgresSaver (every step persists)
+ InMemoryStore (cross-session preferences)
TOOLS: [TavilySearchResults] ← hardcoded
v4: [summarize] → planning → research ⇄ analysis → synthesis → reflection
↑ │
trim + resume replan ◄──┘
+ PostgresSaver (unchanged)
+ InMemoryStore (unchanged)
TOOLS (via MCP):
├── Filesystem Server → read_file, write_file, list_files
├── Web Search Server → search_web
└── Papers DB Server → search_papers, get_paper
↑ Discovered dynamically at startup
Technical Specifications
Stack
| Technology | Version | Use |
|---|---|---|
| Python | 3.11+ | Runtime |
| langchain, langchain-openai | v1.2+ | LLM, prompts |
| langgraph | v1.0+ | StateGraph, checkpointing, store |
| mcp | latest | The SDK for creating MCP servers |
| langchain-mcp-adapters | latest | The MCP → LangChain tools bridge |
| duckduckgo-search | latest | The web search server's backend |
| langgraph-checkpoint-postgres | latest | PostgresSaver (from v3) |
pip install langchain langchain-openai langgraph mcp langchain-mcp-adapters duckduckgo-search langgraph-checkpoint-postgres python-dotenvEnvironment variables
# .env
OPENAI_API_KEY=sk-proj-your-api-key-here
DATABASE_URL=postgresql://agents:agents_secret@localhost:5432/agents_db
RESEARCH_DIR=./research_outputTAVILY_API_KEY is no longer required — the web search uses DuckDuckGo (no API key). If you have it, you can create a server with Tavily as an alternative backend.
v4 Architecture
┌─────────────────────────────────────────────────────────────────────────────┐
│ RESEARCH AGENT v4 (Host + MCP Client) │
│ │
│ ┌───────────┐ ┌──────────┐ ┌──────────┐ ┌───────────┐ │
│ │ SUMMARIZE │───►│ PLANNING │───►│ RESEARCH │◄──►│ ANALYSIS │ │
│ └───────────┘ └──────────┘ └──────────┘ └───────┬────┘ │
│ ▲ │ │ │
│ trim + resume │ ┌────▼─────┐ │
│ │ │SYNTHESIS │ │
│ │ └────┬─────┘ │
│ │ ┌────▼──────┐ │
│ │ │REFLECTION │ │
│ ┌────────┘ └─────┬─────┘ │
│ │ ┌──────────┼──────────┐ │
│ ┌────▼───┐ "replan" "refine" "end" │
│ │ REPLAN │◄────────────┘ │ │ │
│ └────────┘ ▼ ▼ │
│ REFINEMENT END │
│ MEMORY: PostgresSaver + InMemoryStore (unchanged from v3) │
│ │
│ TOOLS: discovered dynamically via MCP ↓ │
└──────────────┬──────────────────┬──────────────────┬────────────────────────┘
│ stdio │ stdio │ stdio
▼ ▼ ▼
┌──────────────────┐ ┌──────────────────┐ ┌──────────────────┐
│ MCP SERVER: │ │ MCP SERVER: │ │ MCP SERVER: │
│ FILESYSTEM │ │ WEB SEARCH │ │ PAPERS DB │
│ │ │ │ │ │
│ Tools: │ │ Tools: │ │ Tools: │
│ • read_file │ │ • search_web │ │ • search_papers │
│ • write_file │ │ │ │ • get_paper │
│ • list_files │ │ Backend: │ │ │
│ │ │ DuckDuckGo │ │ Backend: │
│ Access: │ │ │ │ In-memory DB │
│ ./research_ │ │ │ │ (mock papers) │
│ output/ │ │ │ │ │
└──────────────────┘ └──────────────────┘ └──────────────────┘
Separate process Separate process Separate process
Deps: pathlib Deps: duckduckgo- Deps: no extra
search
Three independent processes. If the papers server goes down, the agent keeps using filesystem and web search. Each server has its own isolated dependencies. Adding a fourth server (e.g. Slack, GitHub) is adding an entry to the configuration — the agent discovers it automatically.
Step 1: Create the Filesystem MCP Server
The filesystem server exposes 3 tools so the agent can read, write, and list research files. Create the file servers/filesystem_server.py:
import os
from pathlib import Path
from datetime import datetime
from mcp.server.fastmcp import FastMCP
mcp = FastMCP("filesystem")
RESEARCH_DIR = Path(os.environ.get("RESEARCH_DIR", "./research_output"))
RESEARCH_DIR.mkdir(parents=True, exist_ok=True)
@mcp.tool()
def read_file(filename: str) -> str:
"""Read a research file's content.
filename: the file's name (e.g.: 'rag-summary.md').
Only reads files inside the research directory.
"""
filepath = RESEARCH_DIR / filename
if not filepath.exists():
return f"Error: file '{filename}' not found in {RESEARCH_DIR}"
if not filepath.resolve().is_relative_to(RESEARCH_DIR.resolve()):
return "Error: access outside the research directory is not allowed"
try:
content = filepath.read_text(encoding="utf-8")
return f"=== {filename} ({len(content)} chars) ===\n{content}"
except Exception as e:
return f"Error reading '{filename}': {e}"
@mcp.tool()
def write_file(filename: str, content: str) -> str:
"""Write content to a research file.
filename: the file's name (e.g.: 'rag-summary.md').
content: the complete text to write.
Overwrites if the file already exists.
"""
filepath = RESEARCH_DIR / filename
if not filepath.resolve().is_relative_to(RESEARCH_DIR.resolve()):
return "Error: writing outside the research directory is not allowed"
try:
filepath.write_text(content, encoding="utf-8")
return f"File '{filename}' written ({len(content)} chars) to {RESEARCH_DIR}"
except Exception as e:
return f"Error writing '{filename}': {e}"
@mcp.tool()
def list_files(extension: str = "") -> str:
"""List the files in the research directory.
extension: filter by extension (e.g.: '.md', '.txt'). Empty = all.
"""
try:
files = sorted(RESEARCH_DIR.iterdir())
if extension:
files = [f for f in files if f.suffix == extension]
files = [f for f in files if f.is_file()]
if not files:
return f"No files{' with extension ' + extension if extension else ''} in {RESEARCH_DIR}"
lines = []
for f in files:
size = f.stat().st_size
modified = datetime.fromtimestamp(f.stat().st_mtime).strftime("%Y-%m-%d %H:%M")
lines.append(f" {f.name} ({size:,} bytes, {modified})")
return f"Files in {RESEARCH_DIR} ({len(files)}):\n" + "\n".join(lines)
except Exception as e:
return f"Error listing files: {e}"
if __name__ == "__main__":
mcp.run(transport="stdio")Verify the server
mcp dev servers/filesystem_server.pyIn the inspector: go to Tools, you'll see read_file, write_file, list_files. Try write_file with {"filename": "test.md", "content": "# Test\nIt works."}, then read_file with {"filename": "test.md"}.
Key points:
- Path traversal protection: It verifies the filepath is inside
RESEARCH_DIRwithis_relative_to. An agent must not be able to read/etc/passwd. - Configurable directory:
RESEARCH_DIRcomes from an env var — the server doesn't hardcode paths. - Descriptive returns: Every response includes metadata (size, path) so the LLM makes better decisions.
Step 2: Create the Web Search MCP Server
The web search server replaces v3's hardcoded TavilySearchResults with an MCP server that uses DuckDuckGo. Create servers/web_search_server.py:
from mcp.server.fastmcp import FastMCP
mcp = FastMCP("web-search")
@mcp.tool()
def search_web(query: str, max_results: int = 5) -> str:
"""Search the internet for current information using DuckDuckGo.
query: a specific search term (e.g.: 'LangGraph vs CrewAI 2025').
max_results: the maximum number of results (1-10).
Returns the title, URL and snippet of each result.
"""
try:
from duckduckgo_search import DDGS
with DDGS() as ddgs:
results = list(ddgs.text(query, max_results=min(max_results, 10)))
if not results:
return f"No results for '{query}'. Try rephrasing the search."
output = f"Results for '{query}' ({len(results)}):\n\n"
for i, r in enumerate(results, 1):
title = r.get("title", "No title")
url = r.get("href", "")
body = r.get("body", "No description")
output += f"{i}. {title}\n URL: {url}\n {body}\n\n"
return output
except ImportError:
return "Error: duckduckgo-search not installed. pip install duckduckgo-search"
except Exception as e:
return f"Search error: {type(e).__name__}: {e}"
if __name__ == "__main__":
mcp.run(transport="stdio")Verify the server
mcp dev servers/web_search_server.pyTry search_web with {"query": "Model Context Protocol MCP 2025"}. Verify it returns real results from DuckDuckGo.
Note: DuckDuckGo requires no API key, which simplifies the setup. If you prefer Tavily for result quality, create an alternative server:
@mcp.tool()
def search_web(query: str, max_results: int = 5) -> str:
"""Search for information using Tavily."""
from tavily import TavilyClient
client = TavilyClient()
results = client.search(query, max_results=max_results)
# ... format the resultsThe point: changing the search backend is changing the server's implementation. The agent never finds out — it keeps calling search_web with the same arguments.
Step 3: Create the Paper Database MCP Server
The paper database server exposes search over a collection of academic papers. In production this would connect to PostgreSQL or a papers API (Semantic Scholar, arXiv). For the project we use in-memory data that demonstrates the pattern. Create servers/papers_server.py:
from mcp.server.fastmcp import FastMCP
mcp = FastMCP("papers-db")
PAPERS_DB = [
{
"id": "vaswani2017",
"title": "Attention Is All You Need",
"authors": "Vaswani, Shazeer, Parmar, Uszkoreit, Jones, Gomez, Kaiser, Polosukhin",
"year": 2017,
"abstract": "The dominant sequence transduction models are based on complex recurrent or convolutional neural networks. We propose a new simple network architecture, the Transformer, based solely on attention mechanisms.",
"citations": 120000,
"topics": ["transformers", "attention", "nlp", "deep learning"],
},
{
"id": "devlin2018",
"title": "BERT: Pre-training of Deep Bidirectional Transformers for Language Understanding",
"authors": "Devlin, Chang, Lee, Toutanova",
"year": 2018,
"abstract": "We introduce BERT, designed to pre-train deep bidirectional representations by jointly conditioning on both left and right context in all layers.",
"citations": 95000,
"topics": ["bert", "pre-training", "nlp", "transformers"],
},
{
"id": "lewis2020",
"title": "Retrieval-Augmented Generation for Knowledge-Intensive NLP Tasks",
"authors": "Lewis, Perez, Piktus, Petroni, Karpukhin, Goyal, Küttler, Lewis, Yih, Rocktäschel, Riedel, Kiela",
"year": 2020,
"abstract": "We explore a general-purpose fine-tuning recipe for retrieval-augmented generation (RAG) — models which combine pre-trained parametric and non-parametric memory.",
"citations": 4500,
"topics": ["rag", "retrieval", "generation", "knowledge"],
},
{
"id": "openai2023",
"title": "GPT-4 Technical Report",
"authors": "OpenAI",
"year": 2023,
"abstract": "We report the development of GPT-4, a large-scale, multimodal model which can accept image and text inputs and produce text outputs.",
"citations": 12000,
"topics": ["gpt-4", "multimodal", "llm", "scaling"],
},
{
"id": "anthropic2024",
"title": "Model Context Protocol Specification",
"authors": "Anthropic",
"year": 2024,
"abstract": "MCP is an open protocol that standardizes how applications provide context to LLMs. It defines a client-server architecture for tool integration.",
"citations": 500,
"topics": ["mcp", "protocol", "tools", "agents"],
},
{
"id": "chase2022",
"title": "LangChain: Building applications with LLMs through composability",
"authors": "Chase, Harrison",
"year": 2022,
"abstract": "LangChain is a framework for developing applications powered by language models, providing composable tools and chains.",
"citations": 3000,
"topics": ["langchain", "llm", "framework", "agents"],
},
{
"id": "yao2022",
"title": "ReAct: Synergizing Reasoning and Acting in Language Models",
"authors": "Yao, Zhao, Yu, Du, Shafran, Narasimhan, Cao",
"year": 2022,
"abstract": "We propose ReAct, a paradigm that synergizes reasoning and acting in language models for general task solving.",
"citations": 2800,
"topics": ["react", "reasoning", "agents", "planning"],
},
{
"id": "wei2022",
"title": "Chain-of-Thought Prompting Elicits Reasoning in Large Language Models",
"authors": "Wei, Wang, Schuurmans, Bosma, Xia, Chi, Le, Zhou",
"year": 2022,
"abstract": "We show that generating a chain of thought — a series of intermediate reasoning steps — significantly improves the ability of large language models to perform complex reasoning.",
"citations": 8500,
"topics": ["chain-of-thought", "reasoning", "prompting", "llm"],
},
]
@mcp.tool()
def search_papers(query: str, max_results: int = 5, min_year: int = 0) -> str:
"""Search academic papers by topic in the database.
query: the search term (e.g.: 'RAG retrieval augmented generation').
max_results: the maximum number of results to return (1-10).
min_year: minimum publication year (0 = no filter).
Searches the title, abstract and topics.
"""
query_lower = query.lower()
scored = []
for paper in PAPERS_DB:
if min_year and paper["year"] < min_year:
continue
score = 0
if query_lower in paper["title"].lower():
score += 3
if query_lower in paper["abstract"].lower():
score += 2
for topic in paper["topics"]:
if query_lower in topic or topic in query_lower:
score += 1
query_words = query_lower.split()
for word in query_words:
if word in paper["title"].lower():
score += 1
if any(word in t for t in paper["topics"]):
score += 1
if score > 0:
scored.append((score, paper))
scored.sort(key=lambda x: x[0], reverse=True)
results = scored[:max_results]
if not results:
return f"No papers for '{query}'. Try broader terms."
output = f"Papers found for '{query}' ({len(results)}):\n\n"
for i, (score, p) in enumerate(results, 1):
output += (
f"{i}. [{p['id']}] {p['title']} ({p['year']})\n"
f" Authors: {p['authors']}\n"
f" Citations: {p['citations']:,}\n"
f" Topics: {', '.join(p['topics'])}\n\n"
)
return output
@mcp.tool()
def get_paper(paper_id: str) -> str:
"""Get a paper's complete details by ID.
paper_id: the paper's identifier (e.g.: 'vaswani2017', 'lewis2020').
Use search_papers first to find valid IDs.
"""
paper = next((p for p in PAPERS_DB if p["id"] == paper_id), None)
if not paper:
valid_ids = [p["id"] for p in PAPERS_DB]
return f"Paper '{paper_id}' not found. Valid IDs: {', '.join(valid_ids)}"
return (
f"=== {paper['title']} ===\n"
f"ID: {paper['id']}\n"
f"Authors: {paper['authors']}\n"
f"Year: {paper['year']}\n"
f"Citations: {paper['citations']:,}\n"
f"Topics: {', '.join(paper['topics'])}\n\n"
f"Abstract:\n{paper['abstract']}\n"
)
if __name__ == "__main__":
mcp.run(transport="stdio")Verify the server
mcp dev servers/papers_server.pyTry search_papers with {"query": "RAG"}. Then get_paper with {"paper_id": "lewis2020"}.
The server has 8 papers covering transformers, RAG, agents, and MCP. In production you'd replace PAPERS_DB with queries to the Semantic Scholar API or your PostgreSQL database. The tools' schema doesn't change — only the implementation.
Step 4: Connect the Agent to the MCP Servers
Now the 3 servers exist as independent programs. This step connects them to the agent using langchain-mcp-adapters, which converts MCP tools into LangChain tools compatible with LangGraph.
Server configuration
from langchain_mcp_adapters.client import MultiServerMCPClient
SERVER_CONFIG = {
"filesystem": {
"command": "python",
"args": ["servers/filesystem_server.py"],
"transport": "stdio",
"env": {
"RESEARCH_DIR": "./research_output",
},
},
"web_search": {
"command": "python",
"args": ["servers/web_search_server.py"],
"transport": "stdio",
},
"papers_db": {
"command": "python",
"args": ["servers/papers_server.py"],
"transport": "stdio",
},
}Each entry defines how to launch the server as a subprocess. MultiServerMCPClient starts them, initializes an MCP session with each one, and discovers their tools automatically.
Tool discovery at startup
async def discover_mcp_tools():
"""Connect to every MCP server and discover its tools."""
async with MultiServerMCPClient(SERVER_CONFIG) as mcp_client:
tools = mcp_client.get_tools()
tool_source = {}
for tool in tools:
server_name = "unknown"
for name in SERVER_CONFIG:
if hasattr(tool, "name") and name in tool.name:
server_name = name
break
tool_source[tool.name] = server_name
print(f"Tools discovered ({len(tools)}):")
for tool in tools:
print(f" • {tool.name}: {tool.description[:60]}...")
return tools, tool_sourceWhen you run this, you'll see something like:
Tools discovered (6):
• read_file: Read a research file's content...
• write_file: Write content to a research file...
• list_files: List the files in the research directory...
• search_web: Search the internet for current information using DuckDuck...
• search_papers: Search academic papers by topic in the databa...
• get_paper: Get a paper's complete details by ID...
6 tools discovered from 3 servers. The agent imported no functions. It wrote not a single @tool. It connected to 3 processes and discovered which tools they offer via a standard protocol.
Binding the tools to the model
from langchain_openai import ChatOpenAI
async def create_model_with_mcp_tools(mcp_client):
"""Create a model with the MCP tools bound."""
tools = mcp_client.get_tools()
model = ChatOpenAI(model="gpt-4.1-mini", temperature=0)
model_with_tools = model.bind_tools(tools)
return model_with_tools, toolsThe tools from mcp_client.get_tools() are already LangChain StructuredTools — directly compatible with bind_tools. You don't need any manual conversion.
Step 5: Update the Graph
Extend the state
The v1, v2, and v3 fields stay unchanged. You only add fields for MCP:
from typing import Annotated, TypedDict, Optional
from langchain_core.messages import BaseMessage
from langgraph.graph.message import add_messages
class ResearchPlan(TypedDict):
main_query: str
sub_questions: list[str]
completed_questions: list[str]
priority_order: list[int]
success_criteria: str
class AgentState(TypedDict):
# v1/v2 fields (unchanged)
messages: Annotated[list[BaseMessage], add_messages]
plan: ResearchPlan
iteration_count: int
max_iterations: int
research_data: list[str]
quality_score: float
final_answer: str
metadata: dict
reflection_notes: list[str]
plan_revisions: int
max_plan_revisions: int
reasoning_trace: list[dict]
# v3: Memory Systems (unchanged)
user_id: str
conversation_summary: str
preferences_loaded: bool
# v4: MCP Integration (new)
available_tools: list[str]
active_mcp_servers: list[str]
tool_source: dictThe research node with MCP tools
The research node now uses tools that come from MCP. The logic is the same — the model decides which tool to invoke — but the tools aren't hardcoded:
from langchain_core.messages import AIMessage, HumanMessage, SystemMessage, ToolMessage
def create_research_node(model_with_tools, tools):
"""A factory that creates the research node with MCP tools."""
tool_map = {tool.name: tool for tool in tools}
def research_node(state: dict) -> dict:
plan = state.get("plan", {})
sub_questions = plan.get("sub_questions", [])
completed = state.get("plan", {}).get("completed_questions", [])
pending = [q for q in sub_questions if q not in completed]
if not pending:
return {}
current_question = pending[0]
iteration = state.get("iteration_count", 0)
available_names = ", ".join(tool_map.keys())
response = model_with_tools.invoke([
SystemMessage(content=(
f"You are an expert researcher. Answer the question "
f"using the available tools.\n\n"
f"AVAILABLE TOOLS: {available_names}\n\n"
f"Question to research: {current_question}\n"
f"Plan context: {plan.get('main_query', '')}\n\n"
f"Use search_web for current information. "
f"Use search_papers for academic papers. "
f"Use read_file to consult previous research."
)),
*state.get("messages", [])[-5:],
HumanMessage(content=current_question),
])
new_data = list(state.get("research_data", []))
messages_out = [response]
if response.tool_calls:
for tc in response.tool_calls:
tool_name = tc["name"]
if tool_name in tool_map:
try:
result = tool_map[tool_name].invoke(tc["args"])
tool_msg = ToolMessage(
content=str(result),
tool_call_id=tc["id"],
)
messages_out.append(tool_msg)
new_data.append(f"[{tool_name}] {str(result)[:500]}")
except Exception as e:
tool_msg = ToolMessage(
content=f"Error in {tool_name}: {e}",
tool_call_id=tc["id"],
)
messages_out.append(tool_msg)
new_completed = list(completed)
if current_question not in new_completed:
new_completed.append(current_question)
updated_plan = {**plan, "completed_questions": new_completed}
return {
"messages": messages_out,
"research_data": new_data,
"iteration_count": iteration + 1,
"plan": updated_plan,
}
return research_nodeThe create_research_node function is a factory: it takes the model with tools bound and the tool map, and returns the node. That lets the tools be discovered outside the node and injected — the node knows nothing about MCP, it just knows it has tools.
A synthesis node that saves to the filesystem
In v3, the final report only existed in the state. In v4, the agent can save it as a file:
def create_synthesis_node(model, tool_map):
"""Synthesis that generates a report and optionally saves it to disk."""
def synthesis_node(state: dict) -> dict:
plan = state.get("plan", {})
research_data = state.get("research_data", [])
context_prefix = _build_context_prefix(state)
combined_data = "\n---\n".join(research_data[-10:])
response = model.invoke([
SystemMessage(content=(
f"{context_prefix}"
"Synthesize all the research data into a structured, "
"comprehensive report. Include sources where possible."
)),
HumanMessage(content=(
f"QUERY: {plan.get('main_query', '')}\n\n"
f"DATA:\n{combined_data}"
)),
])
report = response.content
save_messages = []
if "write_file" in tool_map:
try:
from datetime import datetime
filename = f"research-{datetime.now().strftime('%Y%m%d-%H%M%S')}.md"
file_content = (
f"# {plan.get('main_query', 'Research')}\n\n"
f"**Date:** {datetime.now().strftime('%Y-%m-%d %H:%M')}\n\n"
f"{report}"
)
result = tool_map["write_file"].invoke({
"filename": filename,
"content": file_content,
})
save_messages.append(
AIMessage(content=f"Report saved: {result}")
)
except Exception as e:
save_messages.append(
AIMessage(content=f"Couldn't save the report: {e}")
)
return {
"final_answer": report,
"messages": [AIMessage(content=report)] + save_messages,
}
return synthesis_nodeThe node checks whether write_file is available among the discovered tools. If it is, it saves the report. If it isn't (because the filesystem server isn't connected), the report still exists in the state but doesn't get saved to disk. The node adapts to the available tools.
Build the complete graph
from langgraph.graph import StateGraph, START, END
from langgraph.prebuilt import ToolNode
def should_continue_research(state: dict) -> str:
if state.get("quality_score", 0.0) >= 0.7:
return "synthesis"
if state.get("iteration_count", 0) >= state.get("max_iterations", 3):
return "synthesis"
return "research"
def should_replan_or_finish(state: dict) -> str:
if state.get("quality_score", 0.0) >= 0.7:
return "end"
if state.get("plan_revisions", 0) < state.get("max_plan_revisions", 2):
return "replan"
return "end"
def build_research_agent_v4(model_with_tools, tools, model, tool_map):
"""Build the v4 graph with MCP tools."""
graph = StateGraph(AgentState)
research_fn = create_research_node(model_with_tools, tools)
synthesis_fn = create_synthesis_node(model, tool_map)
graph.add_node("summarize", summarize_node)
graph.add_node("planning", planning_node)
graph.add_node("research", research_fn)
graph.add_node("analysis", analysis_node)
graph.add_node("synthesis", synthesis_fn)
graph.add_node("reflection", reflection_node)
graph.add_node("replan", replan_node)
graph.add_edge(START, "summarize")
graph.add_edge("summarize", "planning")
graph.add_edge("planning", "research")
graph.add_edge("research", "analysis")
graph.add_edge("synthesis", "reflection")
graph.add_edge("replan", "research")
graph.add_conditional_edges(
"analysis", should_continue_research,
{"research": "research", "synthesis": "synthesis"},
)
graph.add_conditional_edges(
"reflection", should_replan_or_finish,
{"end": END, "replan": "replan"},
)
return graphThe summarize_node, planning_node, analysis_node, reflection_node, and replan_node nodes are the same ones from v3. You don't touch them. Only research and synthesis change to use MCP tools instead of hardcoded ones.
The Complete Research Agent v4
The summarize_node, planning_node, analysis_node, reflection_node, replan_node nodes and the _parse_json, _trace, _build_context_prefix helpers are identical to v3 — copy them from the M6 project. The memory functions (save/load_user_preferences, save/load_research_history, extract_and_save_preferences) also stay unchanged.
What's new in v4:
- 3 MCP servers —
filesystem_server.py,web_search_server.py,papers_server.py SERVER_CONFIG— The server configuration for MultiServerMCPClientcreate_research_node(model_with_tools, tools)— The research node with MCP toolscreate_synthesis_node(model, tool_map)— Synthesis that saves reports via MCPbuild_research_agent_v4()— The graph using factory nodes with injected tools- State fields —
available_tools,active_mcp_servers,tool_source
To run it end-to-end:
import asyncio
import os
import json
from dotenv import load_dotenv
from datetime import datetime, timezone
load_dotenv()
from langchain.chat_models import init_chat_model
from langchain_openai import ChatOpenAI
from langchain_core.messages import AIMessage, HumanMessage, SystemMessage, ToolMessage
from langchain_mcp_adapters.client import MultiServerMCPClient
from langgraph.checkpoint.postgres import PostgresSaver
from langgraph.store.memory import InMemoryStore
from langgraph.graph import StateGraph, START, END
model = init_chat_model("openai:gpt-4.1-mini")
store = InMemoryStore()
DB_URI = os.environ.get("DATABASE_URL", "")
# ... (include: AgentState, ResearchPlan, _parse_json, _trace,
# _build_context_prefix, every v3 memory function,
# every v3 node: summarize_node, planning_node,
# analysis_node, reflection_node, replan_node,
# create_research_node, create_synthesis_node,
# build_research_agent_v4, SERVER_CONFIG) ...
async def run_research_v4(
query: str,
user_id: str = "default",
thread_id: str | None = None,
max_iterations: int = 3,
max_plan_revisions: int = 2,
verbose: bool = True,
) -> dict:
if thread_id is None:
thread_id = f"research-{datetime.now().strftime('%Y%m%d-%H%M%S')}"
async with MultiServerMCPClient(SERVER_CONFIG) as mcp_client:
tools = mcp_client.get_tools()
tool_map = {tool.name: tool for tool in tools}
tool_names = list(tool_map.keys())
server_names = list(SERVER_CONFIG.keys())
model_with_tools = model.bind_tools(tools)
if verbose:
print(f"\n{'='*60}")
print(f" Research Agent v4 — MCP Integration")
print(f" Query: {query}")
print(f" User: {user_id} | Thread: {thread_id}")
print(f" MCP Servers: {', '.join(server_names)}")
print(f" Tools discovered: {', '.join(tool_names)}")
print(f"{'='*60}")
graph = build_research_agent_v4(model_with_tools, tools, model, tool_map)
if DB_URI:
with PostgresSaver.from_conn_string(DB_URI) as checkpointer:
checkpointer.setup()
agent = graph.compile(checkpointer=checkpointer, store=store)
result = _execute_agent(
agent, query, user_id, thread_id,
tool_names, server_names, tool_map,
max_iterations, max_plan_revisions, verbose,
)
else:
from langgraph.checkpoint.memory import MemorySaver
checkpointer = MemorySaver()
agent = graph.compile(checkpointer=checkpointer, store=store)
result = _execute_agent(
agent, query, user_id, thread_id,
tool_names, server_names, tool_map,
max_iterations, max_plan_revisions, verbose,
)
return result
def _execute_agent(
agent, query, user_id, thread_id,
tool_names, server_names, tool_map,
max_iterations, max_plan_revisions, verbose,
):
config = {"configurable": {"thread_id": thread_id}}
initial_state = {
"messages": [HumanMessage(content=query)],
"plan": {
"main_query": "", "sub_questions": [],
"completed_questions": [], "priority_order": [],
"success_criteria": "",
},
"iteration_count": 0,
"max_iterations": max_iterations,
"research_data": [],
"quality_score": 0.0,
"final_answer": "",
"metadata": {},
"reflection_notes": [],
"plan_revisions": 0,
"max_plan_revisions": max_plan_revisions,
"reasoning_trace": [],
"user_id": user_id,
"conversation_summary": "",
"preferences_loaded": False,
"available_tools": tool_names,
"active_mcp_servers": server_names,
"tool_source": {name: "mcp" for name in tool_names},
}
result = agent.invoke(initial_state, config)
save_research_history(store, user_id, query, result.get("final_answer", "")[:500])
extract_and_save_preferences(store, user_id, result.get("messages", []))
if verbose:
score = result.get("quality_score", 0.0)
tools_used = len(result.get("research_data", []))
print(f"\n Score: {score:.2f} | Data: {tools_used}")
print(f" Thread: {thread_id}")
from pathlib import Path
research_dir = Path(os.environ.get("RESEARCH_DIR", "./research_output"))
if research_dir.exists():
files = list(research_dir.glob("research-*.md"))
if files:
latest = max(files, key=lambda f: f.stat().st_mtime)
print(f" Report saved: {latest.name}")
return result
# ── Execution ────────────────────────────────────────────
if __name__ == "__main__":
save_user_preferences(store, "mike", {
"output_format": "bullet_points",
"preferred_sources": "academic",
"detail_level": "detailed",
})
result = asyncio.run(run_research_v4(
query="What is RAG and how does it compare to fine-tuning for improving LLMs?",
user_id="mike",
thread_id="demo-v4-001",
))
print(f"\n{result.get('final_answer', 'No answer')[:500]}")Expected output (schematic)
============================================================
Research Agent v4 — MCP Integration
Query: What is RAG and how does it compare to fine-tuning...?
User: mike | Thread: demo-v4-001
MCP Servers: filesystem, web_search, papers_db
Tools discovered: read_file, write_file, list_files, search_web, search_papers, get_paper
============================================================
Plan: 3 sub-questions (RAG definition, fine-tuning, comparison)
Preferences loaded: [output_format, preferred_sources, detail_level]
Research (iter 1): search_papers("RAG") → 3 results
Research (iter 2): search_web("RAG vs fine-tuning 2025") → 5 results
Research (iter 3): get_paper("lewis2020") → the full abstract
Analysis: score=0.75 → Synthesis
Synthesis: report generated → write_file("research-20260313-143022.md")
Reflection: score=0.78 → APPROVED
Score: 0.78 | Data: 6
Thread: demo-v4-001
Report saved: research-20260313-143022.md
Recommended Tests
Test 1: Dynamic tool discovery
async def test_discovery():
async with MultiServerMCPClient(SERVER_CONFIG) as mcp_client:
tools = mcp_client.get_tools()
print(f"Tools: {len(tools)}")
for tool in tools:
print(f" {tool.name}: {tool.description[:80]}")
assert len(tools) >= 6, f"Expected 6+ tools, got {len(tools)}"
tool_names = {t.name for t in tools}
assert "search_web" in tool_names, "search_web missing"
assert "read_file" in tool_names, "read_file missing"
assert "search_papers" in tool_names, "search_papers missing"
print("✓ Discovery OK")
asyncio.run(test_discovery())What it validates: The agent discovers 6 tools from 3 servers without knowing in advance which tools exist.
Test 2: A complete research run with MCP tools
result = asyncio.run(run_research_v4(
query="What is the Model Context Protocol and why does it matter for AI agents?",
user_id="test",
thread_id="mcp-test-001",
))
assert result.get("final_answer"), "No final answer"
assert len(result.get("research_data", [])) > 0, "No research data"
print("✓ Research complete")What it validates: The agent completes a research run using tools from MCP servers instead of hardcoded tools.
Test 3: The report gets saved to the filesystem
from pathlib import Path
result = asyncio.run(run_research_v4(
query="Summary of transformers and attention mechanisms",
user_id="test",
thread_id="fs-test-001",
))
research_dir = Path("./research_output")
md_files = list(research_dir.glob("research-*.md"))
assert len(md_files) > 0, "No report was saved"
latest = max(md_files, key=lambda f: f.stat().st_mtime)
content = latest.read_text()
assert len(content) > 100, "The report is empty or too short"
print(f"✓ Report saved: {latest.name} ({len(content)} chars)")What it validates: The synthesis node uses write_file from the filesystem MCP server to save the report.
Test 4: Adding a tool without changing the agent
# 1. Add this tool to papers_server.py:
# @mcp.tool()
# def count_papers() -> str:
# """Count the total number of papers in the database."""
# return f"Total papers: {len(PAPERS_DB)}"
#
# 2. Without changing the agent's code, run it again:
async def test_new_tool():
async with MultiServerMCPClient(SERVER_CONFIG) as mcp_client:
tools = mcp_client.get_tools()
tool_names = {t.name for t in tools}
assert "count_papers" in tool_names, "The new tool wasn't discovered"
print(f"✓ New tool discovered: count_papers (total tools: {len(tools)})")
asyncio.run(test_new_tool())What it validates: You add a tool to the server and the agent discovers it automatically without changing its code. This is MCP's "wow" moment.
Test 5: Memory and checkpointing still work
save_user_preferences(store, "carlos", {
"output_format": "markdown",
"preferred_sources": "academic",
})
result1 = asyncio.run(run_research_v4(
query="What is BERT?",
user_id="carlos", thread_id="mem-a",
))
result2 = asyncio.run(run_research_v4(
query="How does chain-of-thought work?",
user_id="carlos", thread_id="mem-b",
))
prefs = load_user_preferences(store, "carlos")
history = load_research_history(store, "carlos")
assert prefs.get("output_format") == "markdown"
assert len(history) >= 2
print(f"✓ Preferences: {list(prefs.keys())}")
print(f"✓ Research runs: {len(history)}")What it validates: Long-term memory and research history work just like in v3. MCP didn't break the memory.
Success Criteria
-
The 3 MCP servers start up and expose tools.
mcp devshows each server's tools individually. -
The agent discovers 6+ tools dynamically.
MultiServerMCPClient.get_tools()returns tools from all 3 servers with no hardcoding. -
The research run completes with MCP tools. The agent searches the web, searches papers, and saves the report using tools it discovered over the protocol.
-
The report gets saved to the filesystem. After the research, there's a
.mdfile in./research_output/with the report's content. -
Adding a new tool requires no change to the agent. Add
count_papersto the papers server, re-run, and the agent discovers it. -
Checkpointing and memory are still intact. v3's functions (PostgresSaver, InMemoryStore, summarize_node) work with no modifications.
Checklist
-
servers/filesystem_server.pycreated and working (mcp devshows 3 tools) -
servers/web_search_server.pycreated and working (mcp devshows 1 tool) -
servers/papers_server.pycreated and working (mcp devshows 2 tools) -
pip install mcp langchain-mcp-adapters duckduckgo-search -
SERVER_CONFIGdefines the 3 servers with the stdio transport -
MultiServerMCPClientdiscovers 6+ tools at startup -
create_research_nodeuses MCP tools to research -
create_synthesis_nodeuseswrite_fileto save reports -
AgentStateincludesavailable_tools,active_mcp_servers,tool_source - A complete research run executes with MCP tools
- The report is saved as
.mdin./research_output/ - Adding a tool to the server → the agent discovers it with no changes
- PostgresSaver works (checkpoints persist)
- InMemoryStore works (preferences load)
-
summarize_nodeworks (trimming active)
Common Errors
Error 1: "FileNotFoundError" when launching an MCP server
Symptom: MultiServerMCPClient fails with "No such file or directory" when trying to launch a server.
Cause: The path in SERVER_CONFIG["args"] is relative and doesn't resolve from the current directory.
Solution: Use absolute paths or verify you're running the script from the right directory. You can add "cwd" to the config:
"filesystem": {
"command": "python",
"args": ["filesystem_server.py"],
"transport": "stdio",
"cwd": "/path/to/servers/",
}Error 2: The agent doesn't use the MCP tools (it answers without them)
Symptom: The agent generates answers but never invokes search_web, search_papers, etc.
Cause: The tools weren't bound to the model, or the system prompt doesn't mention the available tools.
Solution: Verify that model.bind_tools(tools) receives the list of discovered tools. Verify that the research node's system prompt lists the tools by name. The LLM needs to know the tools exist to decide to use them.
Error 3: "duckduckgo_search not found" in the web search server
Symptom: The web search tool returns an import error.
Cause: duckduckgo-search isn't installed in the environment where the server runs.
Solution: The server runs as a subprocess — it uses the system's Python. pip install duckduckgo-search in the base environment. If you use venvs, make sure the command in the config points to the right Python.
Error 4: Timeout connecting to the MCP servers
Symptom: MultiServerMCPClient hangs or times out during initialization.
Cause: A server prints to stdout (which interferes with the stdio transport) or has a startup error that isn't reported.
Solution: Test each server individually with mcp dev server.py. Verify there's no print() without file=sys.stderr. If the server has import errors, they run before MCP initializes and the client never receives the handshake.
Error 5: write_file doesn't save the report
Symptom: The research finishes successfully but there's no file in ./research_output/.
Cause: The filesystem server's RESEARCH_DIR doesn't match where you're looking, or create_synthesis_node doesn't find write_file in the tool_map.
Solution: Verify that RESEARCH_DIR in the server's env var and the directory you're checking are the same absolute path. Verify that tool_map is built correctly with {tool.name: tool for tool in tools} and that "write_file" is a key.
Error 6: Tools from different servers collide
Symptom: Two servers expose a tool with the same name and the agent uses the wrong one.
Cause: langchain-mcp-adapters can handle namespacing automatically, but if you have collisions, you need to configure it.
Solution: Rename the tools on the servers to avoid collisions, or use the server's prefix in the tool's name. With MultiServerMCPClient, tools generally keep their original names — make sure each server uses unique names.
Error 7: PostgresSaver fails to compile with MCP tools
Symptom: An error when compiling the graph: the state has new fields PostgresSaver doesn't recognize.
Cause: The existing v3 checkpoints don't have the available_tools, active_mcp_servers, tool_source fields.
Solution: Use a new thread_id. The new state fields only affect new checkpoints — LangGraph handles missing fields in existing checkpoints with TypedDict defaults, but it's cleaner to start a new thread for v4.
Connection to M8
Your Research Agent v4 connects to external tools via MCP — filesystem, web search, papers. It's an individual agent with dynamically expanded capabilities. But it's still one agent doing all the work: it plans, researches, analyzes, synthesizes, reflects, and re-plans.
In Module 8 (Multi-Agent Orchestration):
- The supervisor pattern: A coordinating agent assigns tasks to specialized agents (researcher, analyst, writer), each with its own set of MCP tools
- Handoffs: Agents transfer control to each other with shared context — the researcher passes data to the analyst, the analyst to the writer
- Shared state: Multiple agents work on a shared state, each contributing from its specialty
The transition: "One agent with dynamic tools via MCP (M7) → a team of specialized agents, each with its own MCP servers (M8)." MCP makes multi-agent viable: without MCP, each agent would need its tools hardcoded, and adding tools to the team would mean modifying each agent individually.
Resources
- MCP Python SDK — The official SDK for creating servers and clients
- langchain-mcp-adapters — The MCP → LangChain/LangGraph bridge
- MCP Specification — The complete protocol, architecture, and reference
- MCP Servers Repository — Community servers (filesystem, GitHub, etc.)
- LangGraph Documentation — StateGraph, checkpointing, store
- DuckDuckGo Search Python — A search library with no API key
- Building Effective Agents — Anthropic — Tool integration patterns in agents