★ FEATURED ARTICLE 1. 为什么要在 LangGraph 里接入 MCP1.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 SchemaOpenAI 也是 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: listmessages用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( commandself.command, argsself.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 工具调用结果的缓存与复用有些工具调用是幂等的比如查询类工具同样的参数返回同样的结果。这种工具调用可以加缓存避免重复请求。我用的是内存缓存加 TTLkey 是工具名称加参数哈希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 调试编排逻辑。这个分步调试的方法帮我省了很多时间。
阅读完成 · 觉得有帮助？
