ARTICLE · INTELLIGENCE

战地情报 · 详情页

来自尧图项目组的一线实战观察与深度解析

LangGraph 接入 MCP:实现工具即插即用的 Agent 架构

LangGraph 接入 MCP:实现工具即插即用的 Agent 架构

1. 为什么要在 LangGraph 里接入 MCP

1.1 从“工具写死”到“工具即插即用”的转变

做过 Agent 项目的人大概都有过这种体验:一开始用 LangChain 或者 LangGraph 搭一个能调用工具的智能体,感觉挺爽,工具函数直接写在代码里,@tool装饰器一贴,绑定到模型上就能跑。但项目稍微做大一点,问题就来了——工具越来越多,每个工具都要在代码里注册一遍,换个模型要改,换个工具要改,团队里不同人维护的工具散落在各个文件里,最后变成一锅粥。

我去年接手过一个内部知识助手项目,最初只有三个工具:查文档、查数据库、发通知。半年后膨胀到二十多个工具,代码里光工具注册就写了三百多行,每次加一个新工具都要重新部署整个服务。更头疼的是,有些工具是别的团队维护的,他们想更新工具逻辑,还得等我们这边排期发版。这种耦合度,说白了就是把所有鸡蛋放在一个篮子里,篮子一晃全碎。

MCP 协议的出现,本质上就是来解决这个问题的。MCP 全称 Model Context Protocol,你可以把它理解成 AI 应用和外部工具之间的“USB 接口标准”。以前每个工具都要自己写适配层,现在只要工具方按照 MCP 协议暴露一个服务端点,Agent 这边就能自动发现、自动调用,完全不需要改代码。LangGraph 作为编排层,负责控制流程和状态管理,MCP 负责工具的动态接入,两者结合,才算是真正把“工具即插即用”这件事落地了。

1.2 LangGraph 和 MCP 各自扮演什么角色

很多人容易把这两个东西搞混,觉得都是“让 AI 调工具”的,有什么区别?我用一个生活化的类比来解释:LangGraph 像是一个项目经理,它负责决定“先做什么、再做什么、什么条件下走哪个分支”;MCP 像是一个标准化的外包合同,规定了“甲方怎么提需求、乙方怎么交付结果”。项目经理不需要知道外包团队内部怎么运作,只需要按照合同格式发需求、收结果就行。

具体到技术层面,LangGraph 的核心价值在于状态管理和流程编排。它用图结构来定义 Agent 的行为,节点代表一个操作(比如调用模型、执行工具),边代表流转条件。你可以很方便地实现循环、分支、并行这些复杂逻辑。而 MCP 的核心价值在于工具的动态发现和标准化调用。它定义了一套基于 JSON-RPC 的通信协议,客户端可以列出服务端有哪些工具、每个工具需要什么参数、返回什么结果。

两者结合之后,你的 LangGraph 节点里不再需要硬编码工具函数,而是通过 MCP 客户端去动态获取工具列表,然后根据模型输出的意图去调用对应的 MCP 工具。这样一来,工具的新增、修改、下线都不需要动 LangGraph 的代码,只需要 MCP 服务端那边调整就行。

1.3 适合哪些场景和人群

这套方案特别适合以下几种情况:一是工具数量多且频繁变动的项目,比如企业内部的各种微服务接口;二是多团队协作的场景,工具由不同团队维护,需要解耦;三是需要快速验证不同工具组合的实验性项目,不想每次改工具都重新部署。

如果你是一个刚接触 Agent 开发的初学者,建议先把 LangGraph 的基本概念搞清楚,再来看 MCP 的接入。如果你已经用过 LangChain 的工具调用,但觉得维护成本太高,那这篇文章就是为你写的。下面我会从架构设计、核心实现、实操步骤到踩坑经验,一步步拆解怎么把这两个东西串起来。

2. 整体架构设计与核心思路拆解

2.1 三层架构:编排层、协议层、工具层

我在实际项目中总结出来的架构是这样的:最上面是编排层,用 LangGraph 定义 Agent 的状态机,负责决策和流程控制;中间是协议层,用 MCP 客户端封装工具发现和调用逻辑;最下面是工具层,由各个 MCP 服务端提供具体的工具能力。

这样分层的好处是职责清晰。编排层只关心“什么时候该调工具、调完结果怎么处理”,完全不关心工具是怎么实现的。协议层只关心“怎么按照 MCP 规范发请求、收响应”,不关心业务逻辑。工具层只关心“我这个工具具体做什么”,不关心谁在调用。三层之间通过标准接口通信,任何一层的变化都不会影响其他层。

我试过把 MCP 客户端直接塞进 LangGraph 的节点函数里,代码能跑,但很快就发现两个问题:一是每次调用都要重新建立连接,性能很差;二是错误处理逻辑散落在各个节点里,很难统一管理。后来改成独立的协议层,用连接池管理 MCP 会话,错误统一在协议层处理,编排层拿到的永远是干净的結果或者标准化的异常,代码清爽了很多。

2.2 为什么选择 MCP 而不是自定义工具注册

有人可能会问:我自己写一个工具注册中心不行吗?为什么要用 MCP?这个问题我认真想过。自定义注册中心确实能解决一部分问题,但有几个硬伤:第一,你需要自己定义工具描述格式,每个接入方都要按照你的格式来写,学习成本高;第二,你需要自己实现服务发现、健康检查、版本管理这些基础设施,工作量不小;第三,你的方案只能在自己的系统里用,换个平台就要重写。

MCP 作为一套开放协议,已经把这些东西都定义好了。工具描述用 JSON Schema,通信走 JSON-RPC,服务发现通过标准的list_tools方法。更重要的是,越来越多的工具提供商开始原生支持 MCP,你不需要自己写适配层,直接接进来就能用。这就好比以前每个电器都要配一个专用充电器,现在大家都用 Type-C 接口,一根线走天下。

当然,MCP 也不是银弹。它的协议相对重,对于非常简单、固定的工具场景,直接写函数可能更直接。但只要你预期工具会增长、会变化、会由不同人维护,MCP 的投入产出比就很高。

2.3 状态设计:如何在 LangGraph 中管理 MCP 会话

LangGraph 的状态管理是它的核心特性之一。在接入 MCP 的场景下,状态设计需要额外考虑几个东西:MCP 会话的生命周期、工具列表的缓存、调用结果的暂存。

我的做法是在 LangGraph 的 State 里加三个字段:mcp_session用来保存当前会话对象,available_tools用来缓存从 MCP 服务端拉取的工具列表,tool_results用来暂存工具调用的结果。这里有个细节要注意:mcp_session不应该被序列化,因为它包含网络连接对象,所以在定义 State 的时候要用Annotated标记为不参与检查点存储。

另外,工具列表的缓存策略也很重要。如果每次节点执行都去拉一遍工具列表,网络开销会很大。我的做法是在会话初始化的时候拉一次,然后在 State 里缓存起来。如果 MCP 服务端的工具列表可能动态变化,可以加一个定时刷新或者手动刷新的机制。但大多数场景下,工具列表在会话期间是稳定的,缓存一次就够了。

3. 核心细节解析与实操要点

3.1 MCP 客户端的初始化与连接管理

MCP 客户端初始化是整个流程的第一步,也是最容易出问题的地方。MCP 支持多种传输方式,常见的有标准输入输出(stdio)和 HTTP 两种。stdio 方式适合本地工具,HTTP 方式适合远程服务。我两种都用过,下面分别说一下要点。

stdio 方式的初始化大概是这样的:你需要在代码里指定要启动的 MCP 服务端命令,客户端会以子进程的方式启动它,然后通过标准输入输出进行通信。这种方式的好处是简单直接,不需要额外部署服务。但要注意子进程的生命周期管理,如果客户端异常退出,子进程可能会变成孤儿进程。我的做法是在 LangGraph 的图执行结束时,显式关闭 MCP 客户端,确保子进程被回收。

HTTP 方式的初始化需要指定服务端的 URL。这种方式适合工具已经部署在远程服务器上的场景。需要注意的是网络超时和重试策略。我一般会设置连接超时 5 秒,读取超时 30 秒,重试 2 次。超时时间太短容易误判,太长又会影响用户体验。30 秒的读取超时是因为有些工具执行时间较长,比如涉及大数据量查询或者外部 API 调用的工具。

连接管理方面,我强烈建议使用连接池或者会话复用。每次调用都新建连接的话,在工具调用频繁的场景下性能会非常差。实测下来,复用会话能把工具调用的平均延迟降低 60% 以上。

3.2 工具列表的动态获取与格式化

MCP 服务端通过list_tools方法暴露工具列表,每个工具包含名称、描述、参数 Schema 等信息。拿到这些信息之后,需要转换成 LangGraph 能识别的格式,也就是 OpenAI 风格的 function calling 格式。

这个转换过程有几个坑要注意。第一,MCP 的工具描述可能比较长,直接塞给模型会占用大量 token。我的做法是对描述做截断或者摘要,保留核心信息即可。第二,参数 Schema 的格式可能和 OpenAI 的要求有差异,需要做映射。比如 MCP 用的是 JSON Schema,OpenAI 也是 JSON Schema,但某些字段的命名可能不同。第三,工具名称可能包含特殊字符,需要做 sanitize,否则模型可能无法正确输出。

我写了一个转换函数,核心逻辑是遍历 MCP 工具列表,对每个工具提取 name、description、inputSchema,然后组装成{"type": "function", "function": {...}}的格式。这个函数在会话初始化时执行一次,结果缓存在 State 里。

3.3 在 LangGraph 节点中调用 MCP 工具

LangGraph 节点中调用 MCP 工具的标准流程是这样的:首先从 State 里取出缓存的工具列表,绑定到模型上;然后模型根据用户输入决定是否调用工具、调用哪个工具、传什么参数;接着解析模型的输出,如果是工具调用,就通过 MCP 客户端执行对应的工具;最后把工具结果返回给模型,让模型生成最终回复。

这里的关键点是工具调用的错误处理。MCP 工具调用可能因为各种原因失败:网络问题、参数错误、工具内部异常等。我的做法是在协议层统一捕获异常,转换成标准化的错误对象返回给编排层。编排层根据错误类型决定是重试、降级还是直接告诉用户出错了。千万不要让异常直接抛到 LangGraph 的节点外面,否则整个图执行会中断。

还有一个细节是工具调用的超时控制。有些 MCP 工具可能执行很慢,如果不设超时,整个 Agent 会卡住。我一般会给每个工具调用设置 30 秒的超时,超时后返回一个“工具执行超时”的错误,让模型决定下一步怎么做。

4. 完整实操流程与核心环节实现

4.1 环境准备与依赖安装

先把基础环境搭起来。Python 版本建议 3.10 以上,因为 LangGraph 和 MCP 的 Python SDK 都用到了较新的类型语法。依赖主要有三个:langgraph、langchain-openai(或者你用的其他模型提供商)、mcp。

pip install langgraph langchain-openai mcp

如果你用的是 stdio 方式的 MCP 服务端,还需要确保服务端的可执行文件在 PATH 里,或者用绝对路径指定。我一般会在项目根目录下建一个mcp_servers文件夹,把服务端脚本放在里面,用相对路径引用,这样迁移项目的时候不会出问题。

环境变量方面,至少需要配置模型 API 的 key 和 base URL。如果你用的是兼容 OpenAI 接口的模型服务,配置OPENAI_API_KEY和OPENAI_BASE_URL就行。MCP 服务端如果需要认证,也要在这里配置对应的 token。

4.2 定义 LangGraph 的 State 和节点

State 的定义直接决定了整个图的执行逻辑。我的 State 定义大概是这样的:

from typing import Annotated, TypedDict from langgraph.graph.message import add_messages class AgentState(TypedDict): messages: Annotated[list, add_messages] mcp_session: Annotated[object, "not_serializable"] available_tools: list tool_results: list

messages用add_messages注解,这样每轮对话的消息会自动追加。mcp_session标记为不可序列化,避免 LangGraph 在保存检查点时尝试序列化网络连接对象。available_tools和tool_results就是普通的列表。

节点方面,至少需要三个:agent节点负责调用模型,tool_executor节点负责执行 MCP 工具,should_continue条件边负责判断是否需要继续调用工具。agent节点的逻辑是把available_tools绑定到模型上,然后调用模型生成回复。如果模型输出中包含工具调用,就路由到tool_executor节点;否则直接结束。

4.3 MCP 工具调用的完整代码实现

下面是一个简化但可运行的实现示例。先看 MCP 客户端的封装:

from mcp import ClientSession, StdioServerParameters from mcp.client.stdio import stdio_client class MCPClientWrapper: def __init__(self, command: str, args: list[str]): self.command = command self.args = args self.session = None self._client_context = None async def connect(self): server_params = StdioServerParameters( command=self.command, args=self.args ) self._client_context = stdio_client(server_params) read, write = await self._client_context.__aenter__() self.session = ClientSession(read, write) await self.session.__aenter__() await self.session.initialize() async def list_tools(self): result = await self.session.list_tools() return result.tools async def call_tool(self, name: str, arguments: dict): return await self.session.call_tool(name, arguments) async def close(self): if self.session: await self.session.__aexit__(None, None, None) if self._client_context: await self._client_context.__aexit__(None, None, None)

这个封装类管理了 MCP 会话的完整生命周期。connect方法建立连接并初始化会话,list_tools获取工具列表,call_tool执行具体工具,close清理资源。注意这里用的是异步接口,因为 MCP 的 Python SDK 是异步的。

然后是 LangGraph 节点的实现:

async def agent_node(state: AgentState): tools = state["available_tools"] model_with_tools = model.bind_tools(tools) response = await model_with_tools.ainvoke(state["messages"]) return {"messages": [response]} async def tool_executor_node(state: AgentState): last_message = state["messages"][-1] tool_calls = last_message.tool_calls results = [] for call in tool_calls: try: result = await mcp_client.call_tool(call["name"], call["args"]) results.append({ "tool_call_id": call["id"], "role": "tool", "content": str(result.content) }) except Exception as e: results.append({ "tool_call_id": call["id"], "role": "tool", "content": f"工具执行失败: {str(e)}" }) return {"messages": results, "tool_results": results}

agent_node把工具列表绑定到模型上,让模型决定是否调用工具。tool_executor_node遍历模型输出的工具调用请求,逐个通过 MCP 客户端执行,把结果包装成 LangGraph 能识别的 ToolMessage 格式。注意错误处理:如果工具执行失败,不是直接抛异常,而是把错误信息作为工具结果返回给模型,让模型决定怎么处理。

4.4 图的组装与条件路由

最后把节点和边组装成完整的图:

from langgraph.graph import StateGraph, END def should_continue(state: AgentState): last_message = state["messages"][-1] if hasattr(last_message, "tool_calls") and last_message.tool_calls: return "tools" return END graph = StateGraph(AgentState) graph.add_node("agent", agent_node) graph.add_node("tools", tool_executor_node) graph.set_entry_point("agent") graph.add_conditional_edges("agent", should_continue, {"tools": "tools", END: END}) graph.add_edge("tools", "agent") app = graph.compile()

这个图的结构很清晰:从agent节点开始,如果模型输出包含工具调用,就走到tools节点执行工具,执行完再回到agent节点;如果没有工具调用,直接结束。这是一个标准的 ReAct 循环。

运行的时候,先初始化 MCP 客户端并获取工具列表,然后构造初始 State,调用app.ainvoke就行。记得在 finally 块里关闭 MCP 客户端,避免资源泄漏。

5. 常见问题与排查技巧实录

5.1 工具调用失败的高频原因与排查表

实际跑起来之后,工具调用失败是最常见的问题。我整理了一个排查表,覆盖了大部分场景:

问题现象可能原因排查方法解决方案
模型不调用工具工具描述不清晰检查工具 description 是否准确优化描述,补充使用场景
调用参数错误Schema 定义不严谨对比模型输出和 Schema补充参数约束和示例
连接超时网络问题或服务端未启动检查服务端进程和网络增加超时时间,检查服务端日志
工具执行报错工具内部逻辑异常查看 MCP 服务端日志修复工具逻辑,增加错误处理
结果解析失败返回格式不符合预期打印原始返回内容调整解析逻辑,兼容多种格式

这个表我贴在工位上,每次出问题先对照一遍,能解决八成以上的情况。

5.2 模型输出格式不稳定的处理技巧

模型有时候会输出奇怪的格式,比如工具名称拼错、参数类型不对、甚至编造不存在的工具。我的处理策略是三层防护:第一层是在绑定工具的时候,确保工具名称和 Schema 准确无误;第二层是在解析模型输出的时候,做严格的校验,如果工具名称不在可用列表里,直接返回错误让模型重新生成;第三层是在工具执行层,对参数做类型转换和默认值填充,尽量容忍模型的小错误。

还有一个技巧是在系统提示词里明确告诉模型“只能使用提供的工具,不要编造工具名称”。这个简单的提示能显著降低工具名称错误的概率。我实测下来,加了这句话之后,工具调用成功率从 85% 提升到了 95% 以上。

5.3 性能优化的几个实操心得

性能方面,我踩过最大的坑是每次工具调用都重新建立 MCP 连接。后来改成会话复用之后,延迟直接降了一个数量级。具体做法是在 LangGraph 图执行之前建立好 MCP 会话,把会话对象放在 State 里传递,所有节点共用同一个会话。

另一个优化点是工具列表的缓存。前面提到过,工具列表在会话期间一般不变,拉一次缓存起来就行。如果工具确实会动态变化,可以加一个 TTL 缓存,比如 5 分钟刷新一次。

还有一个容易被忽略的点是并发调用。如果模型一次输出了多个工具调用,可以并发执行这些调用,而不是串行。我用asyncio.gather把多个工具调用并发执行,整体耗时从 3 秒降到了 1 秒左右。但要注意,有些工具可能有副作用或者依赖顺序,这种情况下就不能并发,需要根据业务逻辑判断。

5.4 调试与日志记录的实用方法

调试 MCP 工具调用的时候,日志是关键。我的做法是在 MCP 客户端的call_tool方法里加详细的日志,记录调用的工具名称、参数、耗时、返回结果摘要。这样出问题的时候,一看日志就知道是哪一步卡住了。

另外,LangGraph 本身支持流式输出,可以在每个节点执行完之后打印当前 State 的摘要。我一般会打印messages的长度、最后一条消息的类型、tool_results的数量这些关键信息。这样能快速定位是模型决策出了问题,还是工具执行出了问题。

如果问题比较诡异,可以用 LangGraph 的检查点功能,把每一步的 State 保存下来,然后逐步回放。这个功能在排查复杂流程问题的时候特别好用。

6. 进阶玩法与扩展思路

6.1 多 MCP 服务端的聚合管理

实际项目中,工具往往分散在多个 MCP 服务端上。比如一个服务端提供数据库查询工具,另一个提供文件操作工具,还有一个提供外部 API 调用工具。这时候需要一个聚合层,把多个服务端的工具列表合并起来,统一暴露给模型。

我的做法是维护一个 MCP 客户端列表,每个客户端对应一个服务端。初始化的时候并发连接所有服务端,拉取各自的工具列表,然后合并成一个大的工具列表。调用工具的时候,根据工具名称路由到对应的客户端。这里要注意工具名称冲突的问题,如果两个服务端有同名工具,需要加前缀区分。

聚合管理还有一个好处是故障隔离。如果某个服务端挂了,不影响其他服务端的工具使用。我一般会加一个健康检查机制,定期 ping 各个服务端,发现异常就从可用列表中移除,恢复后再加回来。

6.2 工具调用结果的缓存与复用

有些工具调用是幂等的,比如查询类工具,同样的参数返回同样的结果。这种工具调用可以加缓存,避免重复请求。我用的是内存缓存加 TTL,key 是工具名称加参数哈希,value 是调用结果。TTL 根据业务场景设置,查询类工具一般 5 分钟,配置类工具可以更长。

缓存命中率在重复查询较多的场景下能到 40% 以上,效果还是很明显的。但要注意,有副作用的工具不能加缓存,比如发通知、写数据库这种,每次调用都要真实执行。

6.3 从单 Agent 到多 Agent 协作的演进

当工具数量继续增长,单个 Agent 可能管不过来。这时候可以考虑多 Agent 架构:每个 Agent 负责一组相关的工具,Agent 之间通过消息传递协作。LangGraph 对这种模式支持得很好,你可以定义多个 Agent 节点,用条件边控制它们之间的流转。

比如一个 Agent 负责信息查询,一个 Agent 负责数据处理,一个 Agent 负责结果输出。用户请求先到查询 Agent,查询 Agent 调用 MCP 工具获取数据,然后把数据传给处理 Agent,处理 Agent 加工后再传给输出 Agent。这种架构的扩展性更好,每个 Agent 的职责更单一,也更容易维护。

不过多 Agent 也带来了新的复杂度,比如 Agent 之间的通信协议、状态同步、错误传播等。我的建议是先从单 Agent 开始,等确实遇到瓶颈了再考虑拆分。过早优化反而会增加不必要的复杂度。

6.4 安全性与权限控制的考量

MCP 工具调用涉及外部系统,安全性不能忽视。最基本的做法是在 MCP 服务端做权限校验,每个工具调用都要验证调用方是否有权限。我一般会在 MCP 服务端的初始化参数里传入认证 token,每次调用都带上这个 token,服务端验证通过才执行。

另一个考虑是工具调用的审计日志。所有工具调用都要记录谁在什么时候调用了什么工具、传了什么参数、返回了什么结果。这在排查问题和合规审计的时候非常重要。日志要包含调用链 ID,方便追踪一次完整对话中的所有工具调用。

还有一点是敏感数据的处理。如果工具返回的数据包含敏感信息,需要在返回给模型之前做脱敏处理。我一般会在 MCP 客户端的call_tool方法里加一个后处理钩子,对返回结果做过滤和脱敏。

7. 我个人在实际操作中的几点体会

这套方案我在三个项目里落地过,最大的感受是:解耦带来的收益远超预期。以前加一个工具要改代码、测试、部署,现在只需要在 MCP 服务端加一个工具定义,Agent 这边自动就能发现并使用。有一次业务方临时需要一个新工具,我从写 MCP 服务端到 Agent 能调用,总共花了不到半小时,这在以前是不可想象的。

踩过的坑也不少。最开始没做会话复用,每次工具调用都新建连接,延迟高得离谱。后来改成会话复用,性能问题迎刃而解。还有一次是工具描述写得太模糊,模型总是调用错误的工具,后来把描述改得更具体,加上了使用场景和参数示例,准确率大幅提升。

如果让我给刚入门的同学一个建议,那就是:先把 MCP 协议本身搞清楚,再来看 LangGraph 的集成。很多人一上来就想着怎么把两个东西拼起来,结果遇到问题不知道是 MCP 的问题还是 LangGraph 的问题。把基础打牢,后面会顺很多。

最后分享一个小技巧:在开发阶段,可以写一个简单的 MCP 测试客户端,不经过 LangGraph,直接调用 MCP 工具,验证工具本身是否正常工作。这样能把问题隔离在 MCP 层,排查起来快很多。等 MCP 层没问题了,再接入 LangGraph 调试编排逻辑。这个分步调试的方法帮我省了很多时间。

RELATED READING

延伸阅读

更多一线实战笔记与深度复盘,助您持续精进