5. MCP in LangGraph Agents

Overview

In capsules 03 and 04 you built the two pieces of the MCP protocol: servers that expose tools and clients that discover and invoke them. But so far, the orchestration was manual — you decided which tool to call, with which arguments, and in what order. That isn't an agent. An agent is an LLM that decides for itself which tool to use, when, and how to chain the results.

This capsule connects MCP with LangGraph. The result: an agent that starts up, connects to one or more MCP servers, dynamically discovers which tools exist, and uses them inside its StateGraph to solve tasks — without you hardcoding a single tool into the agent's code. New tools appear on the server → the agent discovers and uses them. That's dynamic tool loading, and it fundamentally changes how you build agents.

The bridge between MCP and LangGraph is the langchain-mcp-adapters package. It converts MCP tools into LangChain tools the agent understands. It takes what you discover with list_tools() and transforms it into objects you can pass to bind_tools(), ToolNode, or create_react_agent.

langchain-mcp-adapters

The bridge between two worlds

MCP defines tools with its own format: name, description, and an inputSchema in JSON Schema. LangChain defines tools with BaseTool — Python objects with name, description, args_schema, and an invoke() method. They're different formats representing the same thing.

langchain-mcp-adapters converts from the MCP format to the LangChain format automatically:

MCP Tool                              LangChain Tool
┌─────────────────────┐              ┌─────────────────────────┐
│ name: "search"      │   adapters   │ name: "search"          │
│ description: "..."  │ ──────────→  │ description: "..."      │
│ inputSchema: {...}  │              │ args_schema: PydanticM.  │
│ (JSON-RPC remote)   │              │ invoke(): calls MCP     │
└─────────────────────┘              └─────────────────────────┘

Installation

pip install langchain-mcp-adapters langgraph langchain-openai

The mcp package (the MCP SDK) gets installed automatically as a dependency.

The two main APIs

load_mcp_tools: direct conversion

from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client
from langchain_mcp_adapters.tools import load_mcp_tools

server_params = StdioServerParameters(
    command="python",
    args=["research_server.py"]
)

async with stdio_client(server_params) as (read, write):
    async with ClientSession(read, write) as session:
        await session.initialize()

        tools = await load_mcp_tools(session)

        for tool in tools:
            print(f"  {tool.name}: {tool.description}")

Internally, load_mcp_tools does: list_tools() → iterate → convert_mcp_tool_to_langchain_tool() on each one. The result is a list of LangChain tool objects — exactly what you pass to bind_tools() or ToolNode.

MultiServerMCPClient: the simple way

from langchain_mcp_adapters.client import MultiServerMCPClient

client = MultiServerMCPClient(
    {
        "research": {
            "command": "python",
            "args": ["research_server.py"],
            "transport": "stdio",
        }
    }
)

tools = await client.get_tools()

A single call. The client opens the connection, initializes the session, discovers the tools, converts them, and returns you the list. For a simple agent, that's all you need.

Connecting an Agent to an MCP Server

The complete example: a ReAct agent with MCP tools

import asyncio
from langchain_mcp_adapters.client import MultiServerMCPClient
from langchain.agents import create_agent
from langchain.chat_models import init_chat_model

async def main():
    model = init_chat_model("openai:gpt-4o")

    client = MultiServerMCPClient(
        {
            "research": {
                "command": "python",
                "args": ["research_server.py"],
                "transport": "stdio",
            }
        }
    )

    tools = await client.get_tools()
    print(f"Tools discovered: {[t.name for t in tools]}")

    agent = create_agent(model, tools)
    response = await agent.ainvoke(
        {"messages": "Search papers about RAG techniques and summarize the results"}
    )

    for msg in response["messages"]:
        print(f"[{msg.type}] {msg.content[:200] if msg.content else '(tool call)'}")

asyncio.run(main())

Expected output:

Tools discovered: ['search_papers', 'get_paper_details', 'summarize_text']
[human] Search papers about RAG techniques and summarize the results
[ai] (tool call)
[tool] Found 3 papers for 'RAG techniques':...
[ai] (tool call)
[tool] Summary: • RAG combines retrieval with generation...
[ai] I found 3 relevant papers about RAG techniques...

The agent discovered search_papers and summarize_text from the MCP server. It decided to use them — search first, then summarize. You didn't write that logic. The LLM decided it based on the descriptions it discovered via MCP.

What happens internally

The flow: MultiServerMCPClient connects via stdio → calls list_tools() → converts each MCP tool to a LangChain tool → create_agent creates a ReAct StateGraph with those tools → the LLM sees the descriptions and decides which to call → each tool's invoke() does a session.call_tool() to the MCP server → the result comes back as a ToolMessage.

The agent doesn't know the tools come from MCP. To it, they're normal LangChain tools.

Dynamic Tool Loading

The core concept

Dynamic tool loading means the agent doesn't know which tools it has until it starts up and connects to the MCP server. The tools aren't hardcoded — they get discovered at runtime:

• You add a tool to the server → the agent has it available the next time it starts
• You remove a tool from the server → the agent stops offering it
• You change a tool's description → the LLM interprets it differently
• You don't touch the agent's code in any of these cases

Before vs. after MCP

Pattern: an agent factory with tool discovery

async def create_mcp_agent(server_configs: dict, model_name: str = "openai:gpt-4o"):
    """Create an agent with tools discovered from MCP servers."""
    model = init_chat_model(model_name)
    client = MultiServerMCPClient(server_configs)
    tools = await client.get_tools()

    if not tools:
        raise ValueError("No tools were discovered from any server")

    print(f"Agent created with {len(tools)} tools: {[t.name for t in tools]}")
    return create_agent(model, tools)

Every time you call create_mcp_agent, it discovers the current tools. If the server changed, the agent reflects the changes.

Tool Refresh at Runtime

Dynamic tool loading at startup is good, but what happens if the MCP server adds tools while the agent is already running? With the basic pattern, the agent only discovers tools once. The solution: periodic re-discovery.

import time
from langchain_mcp_adapters.client import MultiServerMCPClient
from langchain.agents import create_agent
from langchain.chat_models import init_chat_model

class RefreshableAgent:
    """An agent that re-discovers tools periodically."""

    def __init__(self, server_configs: dict, model_name: str = "openai:gpt-4o",
                 refresh_interval: int = 300):
        self.server_configs = server_configs
        self.model = init_chat_model(model_name)
        self.refresh_interval = refresh_interval
        self.agent = None
        self.tool_names: set[str] = set()
        self.last_refresh = 0

    async def _refresh_tools(self):
        client = MultiServerMCPClient(self.server_configs)
        new_tools = await client.get_tools()
        new_names = {t.name for t in new_tools}

        added = new_names - self.tool_names
        removed = self.tool_names - new_names
        if added:
            print(f"[refresh] New tools: {added}")
        if removed:
            print(f"[refresh] Removed tools: {removed}")

        self.tool_names = new_names
        self.agent = create_agent(self.model, new_tools)
        self.last_refresh = time.time()

    async def ainvoke(self, inputs: dict):
        if self.agent is None or (time.time() - self.last_refresh) > self.refresh_interval:
            await self._refresh_tools()
        return await self.agent.ainvoke(inputs)

How it works

Time 0s      → First invocation: discovers tools, creates the agent
Time 60s     → Invocation: uses the existing agent (not expired)
Time 301s    → Invocation: refresh_interval exceeded → re-discovers
               If the server added "cite_paper" → the agent has it

When to refresh vs. restart

The refresh has a cost: it opens connections, does discovery, re-compiles the graph. Don't do it on every invocation unless you need to.

Multi-Server Integration

One agent, multiple tool sources

In production, tools come from different domains. MultiServerMCPClient handles this natively:

async def main():
    model = init_chat_model("openai:gpt-4o")

    client = MultiServerMCPClient(
        {
            "research": {
                "command": "python",
                "args": ["research_server.py"],
                "transport": "stdio",
            },
            "filesystem": {
                "command": "python",
                "args": ["filesystem_server.py"],
                "transport": "stdio",
            },
            "weather": {
                "url": "http://localhost:8000/mcp",
                "transport": "http",
            },
        }
    )

    tools = await client.get_tools()
    print(f"Total tools from 3 servers: {len(tools)}")

    agent = create_agent(model, tools)
    response = await agent.ainvoke(
        {"messages": "Search papers about RAG and save a summary in notes.md"}
    )

The agent receives all the tools combined. The LLM decides which to use based on the descriptions. For that query, it would use search_papers (research) and write_file (filesystem).

Tool namespacing: avoiding collisions

If two servers have a search tool, you get a collision. For explicit namespace control:

from langchain_mcp_adapters.tools import load_mcp_tools

async def load_namespaced_tools(name: str, server_params) -> list:
    async with stdio_client(server_params) as (read, write):
        async with ClientSession(read, write) as session:
            await session.initialize()
            tools = await load_mcp_tools(session)
            for tool in tools:
                tool.name = f"{name}__{tool.name}"
                tool.description = f"[{name}] {tool.description}"
            return tools

research_tools = await load_namespaced_tools("research", StdioServerParameters(...))
fs_tools = await load_namespaced_tools("fs", StdioServerParameters(...))
all_tools = research_tools + fs_tools
# ['research__search_papers', 'fs__read_file', 'fs__write_file']

The server__tool convention (double underscore) is the same one you used in capsule 04. Now the LLM sees research__search_papers and knows it's the research server's search.

Mixing transports

client = MultiServerMCPClient(
    {
        "local_tools": {
            "command": "python",
            "args": ["local_server.py"],
            "transport": "stdio",
        },
        "cloud_api": {
            "url": "https://mcp-api.mycompany.com/mcp",
            "transport": "http",
            "headers": {"Authorization": "Bearer sk-prod-xxx"},
        },
    }
)

The agent neither knows nor cares which transport each server uses. To it, they're all just LangChain tools.

MCP Tools in a Custom StateGraph

Beyond create_react_agent

create_react_agent is convenient, but in earlier modules you learned to build custom StateGraphs with custom nodes, typed states, and conditional edges. MCP tools integrate the same way.

A StateGraph with MCP tools

import asyncio
from langchain_mcp_adapters.client import MultiServerMCPClient
from langgraph.graph import StateGraph, MessagesState, START
from langgraph.prebuilt import ToolNode, tools_condition
from langchain.chat_models import init_chat_model

async def main():
    model = init_chat_model("openai:gpt-4o")

    client = MultiServerMCPClient(
        {"research": {"command": "python", "args": ["research_server.py"], "transport": "stdio"}}
    )
    tools = await client.get_tools()
    model_with_tools = model.bind_tools(tools)

    def call_model(state: MessagesState):
        response = model_with_tools.invoke(state["messages"])
        return {"messages": response}

    builder = StateGraph(MessagesState)
    builder.add_node("agent", call_model)
    builder.add_node("tools", ToolNode(tools))

    builder.add_edge(START, "agent")
    builder.add_conditional_edges("agent", tools_condition)
    builder.add_edge("tools", "agent")

    graph = builder.compile()

    response = await graph.ainvoke(
        {"messages": [("human", "Search papers about self-RAG and give me the details")]}
    )
    for msg in response["messages"]:
        if msg.content:
            print(f"[{msg.type}] {msg.content[:150]}")

asyncio.run(main())

This is identical to how you built graphs in earlier modules — the only difference is where the tools come from.

A StateGraph with custom state and extra nodes

The pattern extends to typed states with extra fields. Add a ResearchState with papers_found: int, a count_papers node that post-processes MCP tool results, and a should_continue that decides whether to keep calling tools or go to the counter:

import operator
from typing import Annotated
from typing_extensions import TypedDict
from langchain_core.messages import AnyMessage, SystemMessage

class ResearchState(TypedDict):
    messages: Annotated[list[AnyMessage], operator.add]
    papers_found: int

async def build_research_graph():
    model = init_chat_model("openai:gpt-4o")
    client = MultiServerMCPClient(
        {"research": {"command": "python", "args": ["research_server.py"], "transport": "stdio"}}
    )
    tools = await client.get_tools()
    model_with_tools = model.bind_tools(tools)

    def agent_node(state: ResearchState):
        system = SystemMessage(content="You are a research assistant. Use the available tools.")
        return {"messages": [model_with_tools.invoke([system] + state["messages"])]}

    def count_papers(state: ResearchState):
        tool_msgs = [m for m in state["messages"] if m.type == "tool"]
        return {"papers_found": sum(1 for m in tool_msgs if "paper" in m.content.lower())}

    def should_continue(state: ResearchState):
        last = state["messages"][-1]
        return "tools" if hasattr(last, "tool_calls") and last.tool_calls else "count"

    builder = StateGraph(ResearchState)
    builder.add_node("agent", agent_node)
    builder.add_node("tools", ToolNode(tools))
    builder.add_node("count", count_papers)
    builder.add_edge(START, "agent")
    builder.add_conditional_edges("agent", should_continue, {"tools": "tools", "count": END})
    builder.add_edge("tools", "agent")
    return builder.compile()

MCP provides the tools, but your graph's logic orchestrates them however you need. The count_papers node is custom logic that doesn't exist in create_react_agent.

Connection to the Project

In this module's project (capsule 08), the Research Agent transforms. It stops having hardcoded tools and connects to 3 MCP servers:

1. Research MCP Server — search_papers, get_paper_details, save_finding
2. Filesystem MCP Server — read_file, write_file, list_directory
3. Web Search MCP Server — Replaces module 3's hardcoded web search tool

┌──────────────────────────────┐
│     Research Agent           │
│     (LangGraph StateGraph)   │
│                              │
│  model_with_tools ──────────────┐
│  ToolNode(all_mcp_tools) ───────┤
└──────────────────────────────┘  │
                                  │
    ┌─────────────────────────────┤
    │         │                   │
    ▼         ▼                   ▼
┌────────┐ ┌──────────┐ ┌──────────────┐
│Research│ │Filesystem│ │  Web Search  │
│ Server │ │  Server  │ │    Server    │
│ (stdio)│ │  (stdio) │ │   (http)    │
└────────┘ └──────────┘ └──────────────┘

In capsule 06 you'll see how to use ecosystem servers (GitHub, Slack, PostgreSQL) instead of building everything yourself. In capsule 07 you'll take this to production with auth, rate limiting and monitoring.

The Research Agent's progression: M2 (hardcoded tools) → M3 (function calling) → M4 (StateGraph) → M5 (planning) → M6 (memory) → M7 (dynamic tools via MCP).

Troubleshooting

Problem 1: "ModuleNotFoundError: No module named 'langchain_mcp_adapters'"

Cause: The package isn't installed, or the wrong virtual environment.

Solution:

pip install langchain-mcp-adapters
python -c "from langchain_mcp_adapters.tools import load_mcp_tools; print('OK')"

Check with which python that it points to the right environment.

Problem 2: "Tools discovered: 0"

Cause: The server doesn't register tools, or the path is wrong.

Solution: Verify the server works: mcp dev research_server.py. If the inspector shows tools, use absolute paths in MultiServerMCPClient:

"args": ["/absolute/path/research_server.py"],

Problem 3: The agent ignores the MCP tools and answers without using them

Cause: Vague descriptions or a system prompt with no usage instructions.

Solution: Improve the docstrings on the server. Add to the system prompt: "ALWAYS use search_papers when the user asks to search. Don't invent data."

Problem 4: "TypeError" when using await with MultiServerMCPClient

Cause: Async code outside an async function, or a version incompatibility.

Solution: All the logic must be inside an async def run with asyncio.run(). Upgrade the packages:

pip install --upgrade langchain-mcp-adapters langgraph langchain-openai

Problem 5: The agent fails on subsequent invocations

Cause: The MCP session closed between invocations.

Solution: With MultiServerMCPClient, each get_tools() opens a new session. If you reuse tools, use the RefreshableAgent pattern that re-creates the client, or client.session(name) as a context manager for persistent sessions.

Exercises

Exercise 1: A basic agent with MCP tools (Easy)

Create an agent that connects to research_server.py via MultiServerMCPClient, discovers the tools, prints what it discovered, and answers "What papers are there about attention mechanisms?".

import asyncio
from langchain_mcp_adapters.client import MultiServerMCPClient
from langchain.agents import create_agent
from langchain.chat_models import init_chat_model

async def main():
    model = init_chat_model("openai:gpt-4o")
    client = MultiServerMCPClient(
        {"research": {"command": "python", "args": ["research_server.py"], "transport": "stdio"}}
    )
    tools = await client.get_tools()

    print(f"=== {len(tools)} tools discovered ===")
    for tool in tools:
        print(f"  {tool.name}: {tool.description[:80]}")

    agent = create_agent(model, tools)
    response = await agent.ainvoke({"messages": "What papers are there about attention mechanisms?"})

    for msg in response["messages"]:
        if msg.content:
            print(f"\n[{msg.type}] {msg.content[:300]}")

asyncio.run(main())

Exercise 2: Multi-server with two servers (Medium)

Create math_server.py (tools: add, multiply) and text_server.py (tools: word_count, to_uppercase). Connect an agent to both and invoke: "Multiply 7 by 8 and convert the result to uppercase text".

import asyncio
from langchain_mcp_adapters.client import MultiServerMCPClient
from langchain.agents import create_agent
from langchain.chat_models import init_chat_model

async def main():
    model = init_chat_model("openai:gpt-4o")
    client = MultiServerMCPClient({
        "math": {"command": "python", "args": ["math_server.py"], "transport": "stdio"},
        "text": {"command": "python", "args": ["text_server.py"], "transport": "stdio"},
    })
    tools = await client.get_tools()
    print(f"Tools from 2 servers: {[t.name for t in tools]}")

    agent = create_agent(model, tools)
    response = await agent.ainvoke(
        {"messages": "Multiply 7 by 8 and convert the result to uppercase text"}
    )
    for msg in response["messages"]:
        if msg.content:
            print(f"[{msg.type}] {msg.content[:200]}")

asyncio.run(main())

Exercise 3: A custom StateGraph with MCP tools (Medium)

Build a manual StateGraph (not create_react_agent) with MessagesState, a call_model node with bind_tools, a ToolNode, and tools_condition. Connect to research_server.py.

import asyncio
from langchain_mcp_adapters.client import MultiServerMCPClient
from langgraph.graph import StateGraph, MessagesState, START
from langgraph.prebuilt import ToolNode, tools_condition
from langchain.chat_models import init_chat_model

async def main():
    model = init_chat_model("openai:gpt-4o")
    client = MultiServerMCPClient(
        {"research": {"command": "python", "args": ["research_server.py"], "transport": "stdio"}}
    )
    tools = await client.get_tools()
    model_with_tools = model.bind_tools(tools)

    def call_model(state: MessagesState):
        return {"messages": model_with_tools.invoke(state["messages"])}

    builder = StateGraph(MessagesState)
    builder.add_node("agent", call_model)
    builder.add_node("tools", ToolNode(tools))
    builder.add_edge(START, "agent")
    builder.add_conditional_edges("agent", tools_condition)
    builder.add_edge("tools", "agent")

    graph = builder.compile()
    response = await graph.ainvoke(
        {"messages": [("human", "Search papers about self-RAG and summarize the findings")]}
    )
    for msg in response["messages"]:
        if msg.content:
            print(f"[{msg.type}] {msg.content[:200]}")

asyncio.run(main())

Exercise 4: RefreshableAgent with change detection (Hard)

Implement a RefreshableAgent that, when it re-discovers tools, compares them with the previous ones and reports: new tools, removed tools, and tools with a changed description. Include a configurable refresh_interval.

import time
from langchain_mcp_adapters.client import MultiServerMCPClient
from langchain.agents import create_agent
from langchain.chat_models import init_chat_model

class RefreshableAgent:
    def __init__(self, server_configs: dict, model_name: str = "openai:gpt-4o",
                 refresh_interval: int = 300):
        self.server_configs = server_configs
        self.model = init_chat_model(model_name)
        self.refresh_interval = refresh_interval
        self.agent = None
        self.last_refresh = 0
        self.tool_snapshot: dict[str, str] = {}

    async def _detect_changes(self, new_tools: list) -> dict:
        new_snapshot = {t.name: t.description for t in new_tools}
        changes = {"added": [], "removed": [], "changed": []}

        for name in new_snapshot:
            if name not in self.tool_snapshot:
                changes["added"].append(name)
            elif new_snapshot[name] != self.tool_snapshot[name]:
                changes["changed"].append(name)
        for name in self.tool_snapshot:
            if name not in new_snapshot:
                changes["removed"].append(name)

        self.tool_snapshot = new_snapshot
        return changes

    async def _refresh_tools(self):
        client = MultiServerMCPClient(self.server_configs)
        new_tools = await client.get_tools()
        changes = await self._detect_changes(new_tools)

        if any(changes.values()):
            print(f"[refresh] Changes:")
            for kind, names in changes.items():
                if names:
                    print(f"  {kind}: {names}")
        else:
            print(f"[refresh] No changes ({len(new_tools)} tools)")

        self.agent = create_agent(self.model, new_tools)
        self.last_refresh = time.time()

    async def ainvoke(self, inputs: dict):
        if self.agent is None or (time.time() - self.last_refresh) > self.refresh_interval:
            await self._refresh_tools()
        return await self.agent.ainvoke(inputs)

Exercise 5: An agent with custom state and a post-processing node (Hard)

Create a StateGraph with a ResearchState (messages + papers_found: int). Include agent, ToolNode, and counter nodes (the counter counts paper results). The counter runs when the agent finishes using tools. Use the pattern from the "StateGraph with custom state" section as the base.

import asyncio, operator
from typing import Annotated
from typing_extensions import TypedDict
from langchain_mcp_adapters.client import MultiServerMCPClient
from langgraph.graph import StateGraph, START, END
from langgraph.prebuilt import ToolNode
from langchain.chat_models import init_chat_model
from langchain_core.messages import AnyMessage, SystemMessage

class ResearchState(TypedDict):
    messages: Annotated[list[AnyMessage], operator.add]
    papers_found: int

async def main():
    model = init_chat_model("openai:gpt-4o")
    client = MultiServerMCPClient(
        {"research": {"command": "python", "args": ["research_server.py"], "transport": "stdio"}}
    )
    tools = await client.get_tools()
    model_with_tools = model.bind_tools(tools)

    def agent_node(state: ResearchState):
        system = SystemMessage(content="You are a research assistant. Use the tools.")
        return {"messages": [model_with_tools.invoke([system] + state["messages"])]}

    def counter(state: ResearchState):
        tool_msgs = [m for m in state["messages"] if m.type == "tool"]
        return {"papers_found": sum(1 for m in tool_msgs if "paper" in m.content.lower())}

    def should_continue(state: ResearchState):
        last = state["messages"][-1]
        return "tools" if hasattr(last, "tool_calls") and last.tool_calls else "counter"

    builder = StateGraph(ResearchState)
    builder.add_node("agent", agent_node)
    builder.add_node("tools", ToolNode(tools))
    builder.add_node("counter", counter)
    builder.add_edge(START, "agent")
    builder.add_conditional_edges("agent", should_continue, {"tools": "tools", "counter": "counter"})
    builder.add_edge("tools", "agent")
    builder.add_edge("counter", END)

    graph = builder.compile()
    result = await graph.ainvoke({"messages": [("human", "Research papers about RAG")], "papers_found": 0})
    print(f"Papers: {result['papers_found']}")

asyncio.run(main())

Summary

In this capsule you integrated MCP with LangGraph — the step that turns MCP from "an interesting protocol" into "agent infrastructure":

• langchain-mcp-adapters is the bridge: it converts MCP tools into LangChain tools with load_mcp_tools() or MultiServerMCPClient
• MultiServerMCPClient handles connections, sessions, and discovery in a single class. It takes server configurations and returns ready-to-use tools
• Dynamic tool loading: the agent discovers tools at runtime. New tools on the server → available with no change to the agent's code
• Tool refresh: RefreshableAgent re-discovers tools periodically. It detects new, removed, or modified tools
• Multi-server: one agent connected to multiple MCP servers with different transports. Tool namespacing avoids collisions
• Custom StateGraph: MCP tools integrate into bind_tools() and ToolNode just like any LangChain tool. Add custom nodes for routing, post-processing, or business logic

Next capsule: The MCP Ecosystem — instead of building every server yourself, you'll connect your agent to community servers (GitHub, Slack, PostgreSQL, filesystem) that already exist.

Additional Resources

1. langchain-mcp-adapters — GitHub — The package's official repository with usage examples
2. LangChain MCP Integration — Official documentation for the MCP integration in LangChain
3. LangGraph — StateGraph — LangGraph documentation for building agent graphs
4. MCP Python SDK — The official MCP SDK for servers and clients
5. langchain-mcp-adapters API Reference — The complete API reference for the adapter package