Using MCP Tools in LangGraph
LangGraph MCP integration lets a LangGraph agent call tools exposed by a Model Context Protocol server instead of hand-writing a Python function for every capability. If you already have an MCP server for your database, filesystem, or internal API, you can load its tools into a LangGraph graph in a few lines of code and let the graph's model decide when to call them. This guide walks through the setup end to end: installing the adapter package, connecting to one or more MCP servers, converting MCP tools into LangChain-compatible tools, and wiring them into both a prebuilt ReAct agent and a custom StateGraph.
Why connect LangGraph to MCP
MCP standardizes how a tool-serving process describes its tools, accepts calls, and returns results. Before MCP, every team building an agent wrote a custom adapter for each external system: one for Postgres, one for Slack, one for internal REST APIs. MCP servers replace that with a single protocol, so a server written once (in any language) can be consumed by any MCP-aware client, including LangGraph.
The practical payoff for a LangGraph project:
- You stop rewriting tool wrappers per project. An MCP server for your ticketing system works the same way whether it's called from LangGraph, a different agent framework, or Claude Desktop.
- Tool schemas come from the server, not from manually maintained Python type hints, so drift between the tool's real behavior and its description is less likely.
- You can swap or add MCP servers without touching your graph's core logic, because tools are loaded dynamically at startup.
The trade-off is an extra process boundary. Each MCP tool call is a round trip to a subprocess or a remote server, which adds latency compared to an in-process Python function. For high-frequency, low-latency tools, a native LangChain tool is still the better choice. For anything that already exists as an MCP server, or that benefits from being reusable across agent frameworks, the adapter approach wins.
The pieces involved
Three libraries do the work:
langchain-mcp-adapters: converts MCP tools, prompts, and resources into LangChain and LangGraph compatible objects.mcp: the official Python SDK for the Model Context Protocol, used under the hood for the client session and transport.langgraph: the graph orchestration library where the converted tools actually get called.
Install all three:
pip install langgraph langchain-mcp-adapters mcp langchain-openaiSwap langchain-openai for whichever chat model provider you use. Nothing here is model-specific; MCP tool calling works with any LangChain chat model that supports tool calling.
Connecting to a single MCP server
The simplest case is one MCP server running as a local subprocess over stdio. Say you have a filesystem MCP server (a common example server that exposes read/write/list tools scoped to a directory).
import asyncio
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="npx",
args=["-y", "@modelcontextprotocol/server-filesystem", "/tmp/workspace"],
)
async def get_tools():
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)
return tools
tools = asyncio.run(get_tools())
print([t.name for t in tools])load_mcp_tools does the conversion: it calls list_tools on the MCP session, then wraps each returned tool definition in a StructuredTool that LangChain and LangGraph both understand. Each wrapped tool's _arun method sends a call_tool request back to the same session when invoked, so calling the tool from your graph transparently becomes an MCP round trip.
One catch with this pattern: the tools are only valid while the async with blocks are open. If you load tools and then close the session before running your graph, tool calls will fail because the underlying subprocess is gone. For a one-shot script this is fine since everything runs inside the same async with. For a long-running service, you want a persistent client, covered below.
Connecting to multiple MCP servers at once
Most real agents need more than one tool source: a filesystem server, a database server, maybe a web search server. langchain-mcp-adapters ships a MultiServerMCPClient for exactly this, and it is the recommended entry point even for a single server because it manages connection lifecycle for you.
import asyncio
from langchain_mcp_adapters.client import MultiServerMCPClient
client = MultiServerMCPClient(
{
"filesystem": {
"command": "npx",
"args": ["-y", "@modelcontextprotocol/server-filesystem", "/tmp/workspace"],
"transport": "stdio",
},
"search": {
"url": "http://localhost:8931/mcp",
"transport": "streamable_http",
},
}
)
async def get_all_tools():
tools = await client.get_tools()
return tools
tools = asyncio.run(get_all_tools())
for t in tools:
print(t.name, "->", t.description[:60])Each entry in the config dict is a named MCP server with its own transport. stdio spawns a local process and talks over its stdin/stdout, which is the right choice for tools that run on the same machine as your agent, such as a filesystem or shell server. streamable_http (or the older sse transport, still supported for backward compatibility) connects to a remote MCP server over HTTP, which is what you want for a shared internal tool server that multiple agents or teams reuse.
client.get_tools() merges the tool lists from every configured server into a single flat list. If two servers happen to expose tools with the same name, keep an eye on collisions, LangChain tools are looked up by name when the model requests a call, so a duplicate name means only one of them is reachable.
Wiring MCP tools into a prebuilt ReAct agent
The fastest way to get a working agent is langgraph.prebuilt.create_react_agent, which builds a standard ReAct-style graph (model node, tool node, conditional loop back to the model) for you.
import asyncio
from langchain_openai import ChatOpenAI
from langgraph.prebuilt import create_react_agent
from langchain_mcp_adapters.client import MultiServerMCPClient
client = MultiServerMCPClient(
{
"filesystem": {
"command": "npx",
"args": ["-y", "@modelcontextprotocol/server-filesystem", "/tmp/workspace"],
"transport": "stdio",
}
}
)
async def main():
tools = await client.get_tools()
model = ChatOpenAI(model="gpt-4.1")
agent = create_react_agent(model, tools)
result = await agent.ainvoke(
{"messages": [{"role": "user", "content": "List the files in the workspace directory"}]}
)
print(result["messages"][-1].content)
asyncio.run(main())create_react_agent binds the tool list to the model with model.bind_tools(tools), then routes the graph so any AIMessage containing tool calls goes to a ToolNode, which executes the matching MCP-backed tool and appends a ToolMessage with the result. The loop continues until the model responds without requesting another tool call. This is the same graph shape you would get with plain LangChain tools; MCP only changes where the tool's implementation lives, not how the graph is structured.
Wiring MCP tools into a custom StateGraph
For anything beyond a single-turn ReAct loop, build the graph by hand with StateGraph so you control routing, add extra nodes (retrieval, validation, human review), or run multiple agents that share a tool set.
import asyncio
from typing import Annotated
from typing_extensions import TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langgraph.prebuilt import ToolNode, tools_condition
from langchain_openai import ChatOpenAI
from langchain_mcp_adapters.client import MultiServerMCPClient
class State(TypedDict):
messages: Annotated[list, add_messages]
client = MultiServerMCPClient(
{
"filesystem": {
"command": "npx",
"args": ["-y", "@modelcontextprotocol/server-filesystem", "/tmp/workspace"],
"transport": "stdio",
}
}
)
async def build_graph():
tools = await client.get_tools()
model = ChatOpenAI(model="gpt-4.1").bind_tools(tools)
def call_model(state: State):
response = model.invoke(state["messages"])
return {"messages": [response]}
builder = StateGraph(State)
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")
return builder.compile()
async def main():
graph = await build_graph()
result = await graph.ainvoke(
{"messages": [{"role": "user", "content": "Read the README.md file in the workspace"}]}
)
print(result["messages"][-1].content)
asyncio.run(main())tools_condition is a prebuilt router: it checks whether the last message has tool calls attached and routes to "tools" if so, or to END otherwise. ToolNode takes the same tool list you loaded from MCP and executes whichever one the model asked for, in parallel if the model requested multiple calls in one turn. Everything downstream of get_tools() behaves exactly like it would with native LangChain tools, because that conversion already happened.
Handling long-lived connections in a server process
The examples above open the MCP connection, use it, and let it close when the script exits. That is wrong for a web service or a long-running LangGraph deployment, where you do not want to spawn a new subprocess for every incoming request.
The fix is to open the session once at process startup and keep it alive for the life of the service, typically with an AsyncExitStack so all your MCP sessions get cleaned up together on shutdown.
from contextlib import AsyncExitStack
from langchain_mcp_adapters.client import MultiServerMCPClient
class MCPToolProvider:
def __init__(self, config: dict):
self.client = MultiServerMCPClient(config)
self._stack = AsyncExitStack()
self.tools = None
async def start(self):
self.tools = await self.client.get_tools()
async def stop(self):
await self._stack.aclose()
provider = MCPToolProvider(
{
"filesystem": {
"command": "npx",
"args": ["-y", "@modelcontextprotocol/server-filesystem", "/tmp/workspace"],
"transport": "stdio",
}
}
)Call provider.start() during your app's startup hook (FastAPI's lifespan, for example) and provider.stop() on shutdown. Build your graph once with provider.tools and reuse the compiled graph across requests instead of recompiling it per call. Recompiling a graph is cheap, but re-establishing MCP subprocess connections on every request is not.
For a remote streamable_http MCP server, connection reuse matters less since there is no subprocess to spawn, but you still want a single long-lived ClientSession rather than opening a new HTTP session per request, to avoid repeated handshake and initialize overhead.
Filtering and renaming tools before binding them
An MCP server often exposes more tools than you want a given agent to have access to. load_mcp_tools and client.get_tools() return everything the server lists, so filter after loading rather than trying to configure the server itself for each agent.
all_tools = await client.get_tools()
readonly_tools = [t for t in all_tools if t.name in {"read_file", "list_directory"}]
model = ChatOpenAI(model="gpt-4.1").bind_tools(readonly_tools)This is a useful safety boundary. If your filesystem MCP server exposes write_file and delete_file alongside read-only tools, a research agent that only needs to read files should never see the write and delete tools in its bound tool list, regardless of how well you prompt it. Filtering at the tool-list level is a hard constraint the model cannot argue its way around; a system prompt instruction to "never delete files" is not.
You can also rename a tool's description before binding, which is worth doing when the MCP server's description was written for a generic client and does not give your model enough context about when to use it in your specific graph.
for t in all_tools:
if t.name == "list_directory":
t.description = "List files in the user's workspace directory. Use this before reading any file to confirm it exists."Debugging tool calls that silently fail
The most common failure mode when wiring MCP into LangGraph is a tool call that returns an error message instead of raising an exception, because ToolNode catches exceptions from tool execution by default and turns them into a ToolMessage with the error text, so the graph keeps running instead of crashing. That is good for resilience but bad for debugging, since a broken MCP connection can look like the model just getting confused.
Two checks catch most of these:
- Run
await session.list_tools()directly against your MCP session before wiring anything into LangGraph, and confirm the tool names and schemas look right. If this call fails, the problem is in your MCP server config, not in LangGraph. - Set
handle_tool_errors=Falseon yourToolNodetemporarily while debugging, so a tool failure raises instead of being swallowed into a message the model has to interpret.
tool_node = ToolNode(tools, handle_tool_errors=False)Turn it back on for production once you have confirmed calls are working, since letting the model see and react to a tool error is usually better UX than crashing the whole graph run.
FAQ
Do I need a separate MCP client for every LangGraph agent in my app? No. Build one MultiServerMCPClient, call get_tools() once, and reuse the resulting tool list (or filtered subsets of it) across as many agents or graphs as you need. Each MCP session is independent of how many LangGraph agents consume its tools.
Can I use MCP tools alongside regular LangChain tools in the same graph? Yes. client.get_tools() returns a plain list of StructuredTool objects, so you can concatenate it with any hand-written LangChain tools before calling bind_tools or building your ToolNode. The graph does not distinguish between an MCP-backed tool and a native one at call time.
What happens if the MCP server process crashes mid-conversation? The next tool call against that session raises a connection error, which ToolNode will convert into an error ToolMessage unless you disabled handle_tool_errors. For production services, add a health check that restarts the MCP subprocess and rebuilds the tool list on failure, rather than assuming a stdio subprocess stays up for the life of your app.
Should I use stdio or streamable_http transport? Use stdio when the MCP server runs on the same host as your LangGraph process, which is the common case for filesystem, shell, or local database tools. Use streamable_http when the MCP server is a separate, possibly shared, service that multiple consumers connect to over the network. sse still works for older MCP servers but streamable_http is the current recommended transport.
Does MCP tool calling work with streaming? Yes. Streaming a LangGraph run with astream_events or astream behaves the same whether the tools are MCP-backed or native, since MCP tool execution happens inside the same ToolNode step either way. The MCP round trip adds latency to that step but does not change how you consume the stream.
How do MCP prompts and resources fit into this, versus tools? langchain-mcp-adapters also exposes helpers to load MCP prompts as LangChain prompt templates and MCP resources as retrievable content, separate from load_mcp_tools. Most LangGraph integrations only need the tools helper, but if your MCP server ships reusable prompt templates, check the adapter package's prompt loading functions instead of re-implementing that logic yourself.
BootcampA 30-day guided bootcamp: build, harden and ship a production autonomous agent from scratch.
AI AgentsUnderstand how AI agents really work: the loop, the tools, the memory, and why most agent projects fail.