LangGraph Agents的MCP：从基础到进阶——服务器、传输机制与适配桥接
LangGraph智能体的MCP：从基础到进阶，涵盖主机、客户端与服务器角色，FastMCP服务器，stdio与HTTP传输方式，langchain-mcp-adapters以及常见陷阱。
先决要求很简单：只需熟悉LangGraph项目的结构以及用@tool装饰函数的概念即可。Model Context Protocol是一种新的基础设施，并非一种看待图的新思路，因此解释会从MCP本身开始，等到相关概念彻底理解后再引入LangGraph。
普通工具的扩展问题
LangGraph中的工具通常是这样开始的：
@tool
def search_web(query: str) -> str:
"""Search the web for current information."""
return f"Search results for: {query}"
在不需要扩展的情况下这没问题。假设五个项目中的五个智能体都需要一个“搜索公司数据库”的工具。按照上述模式，同一个工具就需要在五个代码库中分别编写五次，而且可能还要用五种不同的语言来实现。一旦数据库结构发生变化，就需要进行五次修改。如果试图与正在开发其他类型AI应用的同事共享这个工具——甚至可能根本不在LangGraph平台上——那就会彻底失败，因为那个@tool函数只属于LangChain的Python环境。
这种问题的名称是N×M集成问题。N种工具（数据库、API、文件系统）需要与M种AI应用（LangGraph智能体、同事开发的自定义智能体、Claude Desktop、IDE助手）相连接。如果没有统一的标准，就会产生N×M种定制化的集成方案。
模型上下文协议（MCP）作为一种开放且通用的标准，用于将人工智能应用与各类工具及数据相连。一旦构建为MCP服务器，便可以被任何兼容MCP的应用程序使用，无论是LangGraph智能体、Claude Desktop、Cursor还是自定义智能体，都无需编写额外的集成代码。通常可以用USB-C接口作类比：一个连接器标准，两端均可连接任何兼容设备。
从“什么是MCP”到部署在真实LangGraph智能体中的生产级服务器，这一过程共经历五个阶段，每个阶段都会增添新的理解。
第一阶段：什么是MCP
在编写任何代码之前，先从一张图片开始。酒店礼宾员并不亲自知道如何预订餐厅、叫出租车或购买剧院门票。礼宾员会保存一份值得信赖的当地服务清单，每项服务都有对应的专家。当客人想要预订晚餐时，礼宾员会拨打专门的预订服务电话，用双方都能理解的格式提出请求，然后再将回复转达给客人。
MCP就是那条标准电话线路以及那种标准的请求格式。人工智能应用则相当于礼宾员，而MCP服务器则是诸如餐厅预订系统、出租车公司或票务站之类的专业服务。MCP本身是一种通用语言和连接机制，因此任何礼宾员都能与任何服务进行沟通，无需了解每项服务自身的操作方式。
三种角色
每个MCP系统都恰好包含三个参与方，其他功能都是在此基础上构建的。
- 主机。最终用户实际使用的AI应用：此处为LangGraph智能体，其他地方则为Claude Desktop、Cursor或带有Copilot功能的VS Code。它负责处理对话、判断何时需要使用工具并展示结果。
- 客户端。它存在于主机内部，每个客户端都与恰好一台服务器保持一对一的专用连接。连接了三台服务器的主机内部会运行三个客户端。这种一对一的映射是刻意设置的安全边界，从而确保文件系统服务器的客户端不会意外将数据泄露到数据库服务器。
通信格式：JSON-RPC 2.0
客户端与服务器之间的消息交流采用JSON-RPC 2.0格式，这是一种简单且成熟的格式，包含三种消息类型。
请求用于指示需要执行的操作，并包含一个唯一的id，以便后续能够匹配到对应的回复：
{"jsonrpc": "2.0", "id": 1, "method": "tools/call", "params": {"name": "search_db", "arguments": {"query": "Q3 revenue"}}}
响应用于回复请求。它会携带表示成功的结果或表示失败的错误信息，其id与请求的id相同：
{"jsonrpc": "2.0", "id": 1, "result": {"content": [{"type": "text", "text": "Q3 revenue was $4.2M"}]}}
通知是一种单向消息，不期望收到回复，例如用于告知可用工具列表已发生变化。
由于SDK会处理相关操作，因此很少有人会手动编写原始JSON-RPC格式。不过，了解其底层原理能帮助理解后续许多内容，就如同了解HTTP原理即便不直接编写HTTP代码也能提升REST开发能力一样。
第二级：三种基本操作
服务器通过恰好三种基本操作来展现其功能，而在构建服务器时的几乎所有设计决策都是在这三种操作中做出的选择。
工具是指模型可以执行的操作。工具是大型语言模型在推理过程中可能决定调用的函数，类似于普通LangGraph中的装饰函数。两者的区别在于所在位置：工具位于MCP服务器内部，而非LangGraph代码库中，需通过协议来发现并调用。每当需要执行某些操作时——如计算、数据库查询、发送邮件或编辑文件——都可以使用工具。
资源则提供只读上下文。应用程序可以获取资源以向模型提供背景资料，模型无需像调用工具那样主动决定去使用它。资源通过URI来标识，就如同网页通过URL来标识一样。对于配置文件、文档或模型在决定搜索内容前可能需要了解的数据源列表等只读资料，可使用资源来提供。
提示词是可重复使用的模板。提示词是一种预先写好的结构，用于为用户或主机启动特定的工作流程。与工具不同，提示词通常由用户调用而非模型，因此类似于斜杠命令。它们适用于那些常见且可重复的任务，比如“总结这份文档”或“检查这段代码是否存在安全问题”，在这种场景下，清晰、设计良好的指令比让用户自行措辞更为有效。
当不确定该使用哪种时，可以问问该操作是否会改变任何内容，还是仅提供信息。如果它用于创建、更新、删除、发送或执行操作，那就是工具。如果是模型作为背景信息所需的只读数据，那就是资源。如果是用户明确启动的指令模板，那就是提示词。
第3级：配备FastMCP的服务器
FastMCP 是构建 MCP 服务器的首选 Python 框架。它将该协议中复杂的机制（请求处理、模式生成、生命周期管理）转化为几个装饰器，其思路与 LangChain 函数中的 @tool 装饰器类似。安装方法：
pip install fastmcp
一个包含单一工具的完整服务器信息可容纳在一段简短的文本中：
# ── math_server.py ───────────────────────────────────────────
from fastmcp import FastMCP
# Create the server - this object is the container for everything below
mcp = FastMCP(name="Math Server")
@mcp.tool
def add(a: int, b: int) -> int:
"""Add two numbers together."""
return a + b
@mcp.tool
def multiply(a: int, b: int) -> int:
"""Multiply two numbers together."""
return a * b
if __name__ == "__main__":
mcp.run() # Defaults to stdio transport
那是一个有效且可运行的MCP服务器。FastMCP会读取函数的类型提示来生成输入架构，同时读取文档字符串来生成LLM能够看到的描述。这与普通工具的规则相同，只不过风险更高：模糊的文档字符串会让模型不确定何时调用该工具，而这种模糊性会影响所有连接的程序，而不仅仅是一个智能体。
要添加资源，需要使用带有指定URI的装饰器：
# ── Adding to math_server.py ─────────────────────────────────
@mcp.resource("config://server-info")
def server_info() -> dict:
"""Provides metadata about this server's capabilities."""
return {"version": "1.0", "operations": ["add", "multiply"]}
当客户端请求 config://server-info 时，FastMCP 会调用相应函数并返回结果。该功能仅在接收到请求时才会执行，属于惰性求值，因此不会计算模型从未使用过的任何内容。
有时某个资源应该依赖于特定参数，比如某条具体的记录，而非固定的配置数据：
@mcp.resource("history://{operation_id}")
def get_calculation(operation_id: str) -> dict:
"""Retrieve a past calculation by its ID."""
# In a real server, this would look up a database record
return {"id": operation_id, "result": "..."}
URI中的{operation_id}占位符使得它成为一种资源模板。当请求history://abc123时，FastMCP会自动提取abc123并将其传递给对应函数。
提示语也是以相同方式定义的：
@mcp.prompt
def explain_calculation(operation: str, result: str) -> str:
"""Generate a request to explain a calculation in plain English."""
return f"Explain how the operation '{operation}' produced the result '{result}', in simple terms a beginner could follow."
要运行服务器：
python math_server.py
在连接真实的 AI 应用程序之前，可使用该生态系统附带的基于浏览器的调试工具 MCP Inspector 进行验证。该工具会列出各类工具，允许手动调用它们，并在无需 LLM 参与的情况下检查相关资源：
fastmcp dev math_server.py
这会打开一个本地网页界面，地址通常为 http://127.0.0.1:6274，你可以通过该界面点击各种工具，输入测试参数，并查看来回传输的 JSON-RPC 消息。建议先在这里进行测试。与在 Inspector 中直接调试相比，通过 LLM 的工具调用行为来排查出故障的工具要困难得多。
第 4 级：传输方式
到目前为止，服务器都是使用默认的传输方式运行的。选择哪种传输方式很重要，因为它决定了 LangGraph 代理将如何建立连接。
可以把传输方式视作物理线路，而 JSON-RPC 则是在这条线路上传输的语言。英语无论通过电话还是无线电都能传输：同一种语言，两种不同的传输介质。无论采用何种传输方式，MCP 服务器都使用相同的 JSON-RPC 2.0 协议，但数据的传输方式会有所不同。
stdio，用于本地工具
这是默认的传输方式，也是当调用 mcp.run() 时未提供参数时的行为：
if __name__ == "__main__":
mcp.run() # stdio is the default
通过标准输入输出功能，主机以子进程的形式启动服务器，并通过向服务器的标准输入写入数据以及从标准输出读取数据来与其通信。这一过程不涉及网络，属于同一台机器上的进程间通信。它适用于与主机一同运行的服务器，比如文件系统工具、本地脚本执行器或个人实用程序，也是开发中最简单且最常用的配置方式。
有一个限制非常重要：在标准输入输出模式下，服务器的stdout专门用于传输协议消息，因此任何多余的print()调试语句都会破坏数据流，日志必须写入stderr。
用于远程工具的流式HTTP
对于网络服务器，则需要明确指定传输方式：
if __name__ == "__main__":
mcp.run(transport="streamable-http", host="0.0.0.0", port=8000)
在此模式下，服务器作为独立的Web服务运行。客户端通过网络发送HTTP POST请求与之连接，还可选择使用服务器推送事件来流式接收响应。这使得服务器真正具备远程特性：它可以部署在另一台机器上、云端或负载均衡器之后，同时为多个主机应用提供服务。当需要在一个团队内共享服务器、集中部署，或让其他机器上的主机能够访问时，可使用此模式，包括在服务器上运行的LangGraph代理与另一处负责数据库访问的MCP服务器之间的通信。
一般原则是：对于本地工具、仅有一名用户且配置简单的场景，适合使用标准输入输出方式，这通常是开发和个人使用时的选择；而对于需要共享、有多名用户或多种应用、需进行生产环境部署及身份验证的场景，则应使用可流式传输的HTTP协议。这正是接下来在LangGraph配置中为每台服务器指定类型时所依据的标准。
第5级：连接到LangGraph
这个桥梁工具是一个名为 langchain-mcp-adapters 的包，由LangChain团队维护，旨在让MCP工具能够在LangGraph智能体中使用。它为MCP所起的作用，就如同 @tool 装饰器对普通函数的作用一样：它将外部MCP工具转换为 BaseTool 对象，该对象可直接插入LangGraph的 ToolNode 中，而无需在图结构中进行特殊处理。转换之后，智能体既不会知道该工具来自MCP服务器还是本地编写，也不会对此加以区分。
可通过以下命令安装它：
pip install langchain-mcp-adapters langgraph "langchain[openai]"
核心对象是 MultiServerMCPClient，它能够同时连接多个MCP服务器，并自由组合传输方式：
# ── MODULE 1 (extended): MCP CLIENT CONFIGURATION ───────────
from langchain_mcp_adapters.client import MultiServerMCPClient
client = MultiServerMCPClient({
# A local server, launched as a subprocess via stdio
"math": {
"command": "python",
"args": ["/absolute/path/to/math_server.py"],
"transport": "stdio",
},
# A remote server, already running, connected over HTTP
"company_data": {
"url": "http://localhost:8000/mcp",
"transport": "http",
},
})
每个词典条目对应一台服务器。键值（"math" 或 "company_data"）仅作为标签，用于在内部对不同服务器上的工具进行分类。该条目反映了传输方式的选择：标准输入输出服务器接收 command 和 args，而 HTTP 服务器接收 url。
加载这些工具只需一次调用即可：
# This single call discovers every tool, on every configured server,
# and converts each one into a LangChain-compatible BaseTool
tools = await client.get_tools()
print([t.name for t in tools])
# ['add', 'multiply', 'search_company_db', ...]
这就是整个集成过程。client.get_tools()会向每个已配置的服务器发送MCP的tools/list请求，获取返回的架构信息后将其封装为工具对象，从图结构的角度来看，这些工具对象与手动编写的@tool函数并无区别。
其余的连接方式与普通LangGraph代理相同。MCP只是改变了工具的来源，并未改变图结构的构建方式。完整的模板如下：
# ============================================================
# LANGGRAPH + MCP AGENT — COMPLETE TEMPLATE
# Extends: Part 1 (core structure)
# ============================================================
# ── MODULE 1: IMPORTS & CONFIGURATION ───────────────────────
import asyncio
from langchain_mcp_adapters.client import MultiServerMCPClient
from langchain.chat_models import init_chat_model
from langchain_core.messages import HumanMessage
from langgraph.graph import StateGraph, MessagesState, START, END
from langgraph.prebuilt import ToolNode, tools_condition
from langgraph.checkpoint.memory import MemorySaver
llm = init_chat_model("openai:gpt-4o")
client = MultiServerMCPClient({
"math": {
"command": "python",
"args": ["./math_server.py"],
"transport": "stdio",
},
"company_data": {
"url": "http://localhost:8000/mcp",
"transport": "http",
},
})
# ── MODULE 2: STATE ──────────────────────────────────────────
class State(MessagesState):
pass
# ── MODULE 4: NODES ──────────────────────────────────────────
# (Module 3, Tools, is now sourced from MCP rather than defined here)
def make_agent_node(llm_with_tools):
"""Factory pattern: tools are loaded asynchronously, so the node
closure is built after the async tool-loading step completes."""
def agent_node(state: State) -> dict:
response = llm_with_tools.invoke(state["messages"])
return {"messages": [response]}
return agent_node
# ── MODULE 6: GRAPH ASSEMBLY (inside an async function) ──────
async def build_graph():
tools = await client.get_tools()
llm_with_tools = llm.bind_tools(tools)
graph_builder = StateGraph(State)
graph_builder.add_node("agent", make_agent_node(llm_with_tools))
graph_builder.add_node("tools", ToolNode(tools))
graph_builder.add_edge(START, "agent")
graph_builder.add_conditional_edges("agent", tools_condition)
graph_builder.add_edge("tools", "agent")
return graph_builder.compile(checkpointer=MemorySaver())
# ── MODULE 7: ENTRYPOINT ──────────────────────────────────────
async def main():
graph = await build_graph()
config = {"configurable": {"thread_id": "session-001"}}
response = await graph.ainvoke(
{"messages": [HumanMessage(content="What's (3 + 5) times 12?")]},
config=config
)
print(response["messages"][-1].content)
if __name__ == "__main__":
asyncio.run(main())
一切变为异步
MCP只做了一项实质性调整。由于每次工具调用都涉及网络或子进程的往返操作，即便是通过标准输入输出也是如此，因此通信本质上是异步的。所以必须等待client.get_tools()的返回结果，图结构构建通常放在async def函数中，而编译后的图则需用ainvoke而非invoke来调用。其余部分，如StateGraph、MessagesState、tools_condition和MemorySaver，则保持不变。
再次使用预构建的路由器
在这个模板中，路由功能实际上被完全取消了。它没有使用自定义路由器，而是采用了现成的tools_condition：如果最后一条消息包含工具调用，则跳转至"tools"，否则结束流程。由于在图结构处理这些内容时，无论是MCP工具还是手动编写的工具都表现为BaseTool对象，因此路由器无法区分二者。
第6级：生产环境考虑
一个可运行的演示版本与可部署的系统之间存在一些差异，而这些差异在真实的MCP和LangGraph环境中最为常见。
保持客户端连接处于活跃状态。 MultiServerMCPClient 默认是无状态的：每次工具调用都会建立新的连接，执行完成后立即断开。对于单个请求来说这没有问题。但对于需要处理大量请求的应用程序，每次调用都开启新的子进程或HTTP连接会带来显著的延迟。在生产环境中，应在进程启动时仅创建一次客户端实例，并在整个应用程序运行期间重复使用它：
# Hold the connection open across the application's lifetime,
# rather than opening/closing per request
async with MultiServerMCPClient({...}) as client:
tools = await client.get_tools()
graph = await build_graph_with(tools)
# ... serve many requests using this same graph ...
避免在Web服务器中使用stdio。 Stio是为本地、单用户、同一台机器上的场景设计的，比如桌面应用启动辅助程序。而当LangGraph代理作为面向众多用户的Web服务部署时，stdio服务器会带来诸多问题：每次请求都会生成新的子进程，且没有自然的方式来共享状态或实现扩展。对于任何服务器端部署，建议使用Streamable HTTP，并诚实地思考是否真的需要单独的MCP进程。在某些情况下，直接调用内部库的普通@tool函数，无论在简洁性还是速度上，都比通过协议进行额外处理更优。
为敏感工具添加人工审批机制。一旦MCP工具被接入系统图，它们就变成了普通工具，因此所有现有的规则都适用。如果服务器执行了某些敏感操作，比如发送邮件、修改数据库或进行资金操作，就需要用与处理其他工具调用时相同的interrupt()审核步骤将其包裹起来。MCP仅能实现工具的发现与标准化，不会默认提供安全性保障，这仍是开发人员需要负责的部分。
服务器无法查看图表状态。MCP服务器作为独立的进程运行，它们无法访问LangGraph的state、检查点或任何内存中的Python对象。当某个调用需要根据存储的用户偏好进行个性化处理时，相关值必须作为工具参数显式传递，或者通过拦截器提供——langchain-mcp-adapters提供了这种高级钩子功能，能够利用图表的运行时上下文在调用发送前对其进行检查和修改。对于大多数智能体而言，将值作为参数传递就已足够，且更为简单。
第7级：多智能体系统中的MCP
MCP与监督者模式相结合使用。多个专业智能体各自从专用的服务器获取相应的工具：
# ── Specialist agents, each pulling tools from a DIFFERENT MCP server ──
async def build_multi_agent_system():
client = MultiServerMCPClient({
"research": {"url": "http://localhost:8001/mcp", "transport": "http"},
"code": {"url": "http://localhost:8002/mcp", "transport": "http"},
})
research_tools = await client.get_tools(server_name="research")
code_tools = await client.get_tools(server_name="code")
researcher_llm = llm.bind_tools(research_tools)
coder_llm = llm.bind_tools(code_tools)
# From here, wire researcher_node, coder_node, and a supervisor_node
# exactly as shown in Part 4 - the rest of the supervisor pattern
# is completely unchanged.
...
MCP并未引入新的竞争架构。由MCP服务器支持的专用组件在结构上与那些由手动编写的工具支持的组件完全相同：调度器以相同方式处理请求，各专用组件也以相同方式反馈信息。MCP仅改变了工具定义的存储位置以及可重用这些工具的应用程序数量。
快速参考
- 架构。 主机是指用户直接使用的AI应用程序（此处为LangGraph代理）。客户端存在于主机中，每个连接的服务器都有一个，负责管理一对一的会话。服务器则是独立的程序，用于提供工具、资源及提示语，且无需知晓有哪些主机正在连接。所有交互——包括请求、响应和通知——都采用JSON-RPC 2.0作为消息格式。
@mcp.tool构建。资源是通过URI访问的只读数据，由应用程序或用户控制，通过@mcp.resource("uri://...")构建。提示语是用户可重复使用的指令模板，通过@mcp.prompt构建。FastMCP(name="...")用于创建服务器，即所有功能的容器。@mcp.tool用于定义函数，可读取模式的类型提示以及用于描述的文档字符串。@mcp.resource("scheme://{param}")用于声明只读数据，而{param}则用于创建模板。@mcp.prompt用于声明提示模板。mcp.run()用于启动服务器，默认通过标准输入输出，如需网络使用则可设置transport="streamable-http"。fastmcp dev server.py可在无需大型语言模型的情况下打开调试工具进行测试。stdio 基于子进程且为本地运行模式，由主机启动服务器，调试输出绝不能发送到 stdout。而 streamable-http 为网络型架构，支持多个并发主机，是共享服务器环境中的默认生产级选择。langchain_mcp_adapters.client 的 MultiServerMCPClient({...}) 可以通过同一个字典，以多种传输方式连接一个或多个服务器。await client.get_tools() 会查找所有工具并将其转换为 BaseTool 类型。tools_condition 是预构建的路由器，对手动编写的工具和通过 MCP 获取的工具都能实现相同的功能。所有操作均为异步处理：get_tools()、图构建以及 ainvoke 都需要使用 await。这里真的需要 MCP 吗？
- 一个工具、一个智能体、一个代码库。无需MCP。使用普通的
@tool函数更为简单，且没有协议开销。 - 可在不同智能体或项目中重复使用的工具，或与使用其他框架的同事共享的工具。可将其构建为MCP服务器，这正是其核心价值所在。
- 希望用现成的数百个社区服务器（如GitHub、Slack、Postgres、Google Drive、Stripe等）来替代手动编写的集成代码。通过
MultiServerMCPClient进行连接，从而省去编写集成代码的步骤，这往往是MCP带来的最大效率提升。 - 位于智能体所在机器之外的工具，或需要同时被多台主机共享的工具。可使用Streamable HTTP作为独立服务。
标准而非替代品
学习新协议总会让人担心它会取代已有的技术。但MCP并非如此。它不会影响LangGraph的现有工具、节点和图结构。变化的是这些工具的来源以及能够共享它们的应用数量。当MCP工具被纳入ToolNode后，它的行为就如同任何手动编写的工具，需遵循相同的interrupt()机制，并适配同样的管理架构。
其发展过程如同阶梯一般：从概念与角色、三种基本元素，到可运行的服务器、传输方式的选择、LangGraph桥接层，再到生产环境优化以及多智能体系统的整合。每一阶段都在稳定的基础之上增添新的理念。结合规范的图结构、内存管理、人工干预的安全机制、多智能体协调与检索功能，这些已涵盖了一个成熟的商用LangGraph应用所需的大部分功能，而MCP则使得相关工具能够超越单一代码库的限制实现一定程度的扩展。
相关阅读
- 实用笔记：为何没人谈论 LangChain、LangGraph 和 AutoAgent — 该实用笔记的详细操作指南：为采用此类架构的团队提供代码模板、检查清单以及可直接使用的代码片段。
- LangGraph 中的 ReAct 模式：从手动循环到 ToolNode 与 StateGraph — 分两步构建 ReAct 智能体：首先通过手动异步循环实现，然后再利用 LangGraph 的 ToolNode、tools_condition 以及 StateGraph 自动处理连续的工具调用。