MCP 这个词最近半年在 AI 应用开发圈子里出现的频率越来越高。我一开始接触的时候也觉得它不过是又一层封装直到有一次在 LangGraph 里真正接入第二个 MCP Server 时才发现协议握手、工具发现、能力协商这些细节如果不搞清楚连个朴素的多服务调用都可能卡上两三天。这篇东西我想从 MCP 协议最早的那次 initialize 握手讲起一直聊到怎么在 LangGraph 里挂载多个 Server把过程中踩过的坑和最终跑通的方案一起整理出来给正在做 Agent 工具层集成的朋友做个参考。这篇文章适合两类人一类是已经把 MCP 单个 Server 的 demo 跑通、但还没想清楚多 Server 怎么编排的开发者另一类是刚接触 LangGraph想知道怎么把 MCP 工具作为一种相对标准化的资源接入图编排的人。我不打算堆概念尽量按照实际调用链路来讲——从握手到工具发现从封装到图执行最后是真实环境里最折磨人的几个问题。1. MCP为什么忽然成了Agent工具层绕不开的环节1.1 它解决的是“每个模型都要重新接一轮工具”的重复劳动在 MCP 出现之前想让一个大模型调用外部服务通常要自己写一套工具封装定义参数格式、处理鉴权、约定错误码、再手动把工具描述塞进提示词里。换一个模型提示词格式变了工具描述的写法也得跟着变换一个服务又要重新走一遍协议设计。这种重复劳动在单工具场景下还能忍一旦 Agent 要同时面对订单查询、库存检查、知识库检索这类多个后端服务光协议适配就能占掉大半开发时间。MCPModel Context Protocol做的事情说直白点就是把这些东西标准化成一个模型与工具之间的统一会话格式。服务端只需要实现一套协议接口客户端统一用同样的消息格式去请求模型侧看到的是统一的工具视图参数说明走 JSON Schema调用方式走统一的请求响应。这样一来工具从专线对接变成了插拔式资源这也是 LangGraph 这类编排框架愿意优先支持它的原因。1.2 单Server演示容易多Server才是真实世界的常态我见过不少项目演示时只挂了一个 MCP Server把某个数据库查询包进去就跑通了。但真实业务很少只有一个数据源。一个稍完整的 Agent 可能要同时触达订单系统、商品库存、物流状态、内部知识库甚至还有外部搜索服务。把这些工具分别暴露成多个 MCP Server让 LangGraph 里的 Agent 根据用户问题自动选择合适的工具是一种很自然的架构演进。这里有一个容易忽略的点多个 Server 意味着多次连接、多套能力协商、多份工具列表而 LangGraph 的 ToolNode 本质上只认一组工具不会自动帮你区分哪个工具来自哪个 Server。所以多 Server 的难点并不在协议本身而在如何把多套连接和工具列表聚合成图里可用的一份资源。这个聚合过程和编排策略是这篇文章后续的核心内容。2. 协议握手不只是打个招呼initialize、能力协商与工具发现2.1 initialize请求里到底装了什么MCP 的会话建立在 JSON-RPC 2.0 消息之上。客户端与 Server 建立传输层连接——常见的有 stdio 和 SSE 两种——之后的第一条消息就是 initialize 请求。这个请求不是简单报个到它要同时携带三样东西协议版本号、客户端标识、客户端能力声明。我的经验是协议版本号protocolVersion最容易被人忽视。SDK 有自带的版本常量但如果客户端和服务端版本不匹配通常不会直接报错而是进入版本协商流程服务端会在响应里返回它支持的版本客户端需要判断是否继续。我踩过一次坑客户端 SDK 升级后自动带上了新版本号而 Server 是旧版本实现的结果某些高级工具调用行为不一致报错还很模糊。握手请求的服务端响应同样包含三个部分Server 标识、能力声明、以及一段可选的服务端说明。这个说明信息有时候会被忽略但它在调试时很有价值某些 Server 会把加载状态、特殊配置提示放在这里。2.2 capabilities协商决定你能否安全调用远端工具initialize 阶段最重要的产出是能力协商结果。比如 Server 是否支持资源订阅是否支持工具调用是否允许客户端发起采样请求这些能力都会明确列在 capabilities 字段里。实践中最关键的是搞清楚采样这个能力的语义。MCP 描述了一个角色反转场景当客户端执行服务端提供的某个工具时服务端反过来可以请求客户端调用一次大模型补全用于生成下一步参数。这个设计很巧妙但也意味着安全边界发生了变化。我建议在 LangGraph 集成场景里默认关闭客户端侧采样能力支持除非你明确知道自己在做什么否则会带来不受控的模型调用开销。能力协商还有一个实际用途告诉客户端能不能用日志通知之类的高级特性。如果 Server 没声明日志能力你发日志订阅请求也没用。所以我在封装客户端时会把 capabilities 解析结果暴露成只读属性后面做功能开关判断都用它。2.3 tools/list是握手之后最关键的环节握手完成之后客户端会发 tools/list 请求获取工具列表。这是整个 MCP 会话里对 Agent 最核心的一次交互因为它决定了模型的工具视图。返回结果里每个工具有三个字段name、description、inputSchema。description 和 inputSchema 的质量直接影响模型选工具的正确率。我给 MCP Server 写工具描述有个简单标准description 里必须说清楚这个工具是干什么的、输入的边界是什么、什么情况下不应该用。比如一个查天气的工具如果 description 只写get_weather模型在涉及是否需要带伞这类意图时就很容易选错。反过来inputSchema 里对必填参数和可选参数的约束越明确模型生成的参数就越不容易跑偏。tools/list 支持分页如果 Server 工具很多会返回 nextCursor需要客户端循环拉取。这个细节在我第一次写客户端时也漏了后来工具数量超过几十个才发现使用 cursor 翻页的必要性。3. LangGraph接入多Server两种路线与我的选型理由3.1 客户端聚合和网关聚合各有什么优劣多 Server 接入本身的架构选择通常先想到两条路线。一条是客户端聚合也就是在 LangGraph 进程内直接创建多个 MCP 客户端连接把它们各自的工具列表拉回来合并成一个统一的工具集传给图里的 ToolNode。另一条是网关聚合做法是引入一个 MCP 聚合网关每个上游服务依然是独立 Server但网关对外暴露成一个统一的超级 ServerLangGraph 只连接网关这一条通道。两条路线的差异很明显。客户端聚合胜在实现简单、链路短每个 Server 的连接生命周期由 Agent 进程直接管理排查问题直观但代价是 LangGraph 进程和服务端之间会有多条长连接连接数量一多资源开销和故障面也会成倍增加。网关聚合则刚好相反下游客户端只维护一条连接服务发现和鉴权都集中在网关一层对于服务数量很多、团队协作复杂的环境更合适但它自己成为了一个必须保证高可用的新组件而且协议转发过程中若实现不严谨可能出现工具参数透传失真。3.2 为什么我最终选择在LangGraph侧做统一聚合我实际的项目规模是两到三个 MCP Server它们分别对接内部数据查询服务和外部检索服务。这个场景下引入网关属于过度设计——又多了一个部署单元和潜在故障点维护成本远大于收益。最终我选择在客户端做聚合把所有 Server 的工具列表合并后交给 LangGraph 去调度。这样做还有一个附带好处我可以在聚合层对工具做重命名和过滤从源头上规避下一节要讲的工具命名冲突。LangGraph 侧只要拿到一个规范化的工具列表后面编排逻辑就保持得很干净。如果你的服务数量超过五个团队协作也需要统一入口我建议再认真评估网关方案不要照搬我这套做法。4. 手把手把两个MCP Server接进LangGraph4.1 环境准备依赖版本选对才能少踩坑先列一下我跑通这套方案用到的关键依赖版本它们之间是兼容的Python 3.11mcp 客户端库用 stdio 传输的 async 版本langgraph0.2.x 以上langchain-core用于 Tool 类型一个容易被忽视的细节是MCP 的客户端依赖在异步环境下的行为差异很大。建议从一开始就写 async 客户端因为 LangGraph 的 ToolNode 天然支持 async 工具调用后续做并行多个 Server 请求会更顺手。4.2 把MCP客户端封装成LangGraph工具核心是把一个 MCP 连接封装成语义上独立的工具集合。封装时要考虑两点工具名的唯一性以及调用函数与 MCP Server 的连接绑定关系。我在聚合时给每个 Server 的工具名加了前缀用 Server 名称加下划线开头作为命名空间这样两个 Server 即使暴露出同名工具也能在 LangGraph 中稳定区分。class McpClientWrapper: def __init__(self, server_name: str, session: ClientSession): self.server_name server_name self.session session async def list_tools(self) - list[dict]: result await self.session.list_tools() tools [] for item in result.tools: tools.append({ name: f{self.server_name}_{item.name}, description: item.description, inputSchema: item.inputSchema, }) return tools async def call_tool(self, mapped_name: str, arguments: dict) - str: raw_name mapped_name[len(self.server_name) 1:] result await self.session.call_tool(raw_name, arguments) return _format_result(result)这里我特别说一下 _format_result 的处理。MCP 工具返回的 content 是列表结构每一项可能是 text 文本也可能是 image 或 resource 类型。我给 LangGraph 用的工具统一转成纯文本把非文本类型的资源用 JSON 序列化后塞回去保证模型拿到的是一个可读的字符串。直接返回原始结构会让模型解析起来很吃力。4.3 多Server编排与工具调用的完整结构有了封装多 Server 的接入逻辑就变成初始化连接池然后合并工具列表再构建图。class MCPManager: def __init__(self, server_configs: dict, connection_factory): self.server_configs server_configs self.connection_factory connection_factory # 根据配置建立 ClientSession async def start(self): self.sessions {} for name, config in self.server_configs.items(): session await self.connection_factory(config) await session.initialize() self.sessions[name] session tool_groups await asyncio.gather( *(McpClientWrapper(name, s).list_tools() for name, s in self.sessions.items()) ) self.all_tools [tool for group in tool_groups for tool in group] async def call_tool(self, mapped_name: str, arguments: dict): server_name mapped_name.split(_, 1)[0] wrapper McpClientWrapper(server_name, self.sessions[server_name]) return await wrapper.call_tool(mapped_name, arguments)这里把 Session 生命周期管理和工具注册解耦了。LangGraph 图构建时我把 all_tools 交给 ToolNode同时构建一个简单的映射函数让 ToolNode 执行工具时找到正确的 Server 会话。多 Server 的并行初始化用 asyncio.gather 来完成比逐个启动快不少。下面是一个 StateGraph 的最小示例它说明了一个 Agent 如何根据用户问题自动选择来自不同 Server 的工具from langgraph.graph import StateGraph, MessageGraph, END from langgraph.prebuilt import ToolNode, tools_condition from typing import TypedDict, Annotated class AgentState(TypedDict): messages: Annotated[list, messages] graph_builder StateGraph(AgentState) tool_node ToolNode(mcp_manager.all_tools, handle_tool_errorsTrue) graph_builder.add_node(tools, tool_node) graph_builder.add_node(agent, call_model_node) graph_builder.add_conditional_edges(agent, tools_condition) graph_builder.add_edge(tools, agent) graph_builder.set_entry_point(agent) graph graph_builder.compile()call_model_node 内部做的事情是把当前对话历史和工具列表拼成消息让模型决定调用哪个工具tools_condition 会检查模型输出里有没有 tool_calls有就跳到 tools 节点执行没有就走到 END。框架层面的这一套大家应该比较熟了重点在于这里传给 ToolNode 的工具是来自不同 MCP Server 的混合集因此工具名必须唯一工具名映射函数必须绑定到正确的 Server 会话。4.4 关于图编排中消息格式的提示一个让我调试了挺久的细节是MCP 工具调用结果返回给模型时必须以 ToolMessage 的形式回填到对话历史里。ToolNode 默认会做这个转换但它要求工具返回值是字符串或可序列化对象。所以我强烈建议在封装层就统一把 MCP 返回的 content 结构转成字符串避免图执行中途出现格式异常。5. 跑通之后才遇到的三个坑工具冲突、超时与重连5.1 同名工具被后注册的Server覆盖这是多 Server 接入中最容易碰到的坑。两个 Server 暴露了同名工具比如都叫 query如果直接在聚合列表里 appendLangGraph 构建工具表时后注册的会覆盖先注册的而模型看到的工具描述还是两份——于是出现一种很诡异的现象模型明明按描述选择了某个 Server 的工具实际执行时却被另一个 Server 的函数接管。我的解决方案就是前面提到的命名空间前缀。在一个内部项目里业务侧希望工具名保持简洁我就改在 description 开头加一行此工具来自某某服务匹配实际函数时仍走前缀映射。这样做既保住了模型侧的可读性也避免了底层调用冲突。两种做法都值得一试关键是不能只改名不复核 description否则模型还是会选错。5.2 长耗时任务把整个Agent卡死的处理MCP 的调用默认没有超时语义如果某个 Server 的工具执行了一个耗时的外部请求图会一直等着。这个问题的麻烦之处在于它不像普通 HTTP 请求那样有明确的超时配置位置超时要由客户端自己去控制。我的处理方式是在封装 call_tool 时用 asyncio.wait_for 包一层并给不同 Server 设置不同的超时阈值。比如内部查询快速失败设为 10 秒外部检索可以宽松到 60 秒。同时把超时报错文本改写成工具调用超时请提醒用户稍后重试让模型能够把这个信息组织到最终回复里而不是让 Agent 直接中断。这里有个取舍超时定太短长任务必然失败定太长Agent 的响应延迟不可接受。比较实用的思路是区分工具类型给阈值而不是一刀切。我还尝试过把长任务拆成发起轮询两个工具把耗时从单次调用里挪出去效果也不错只是需要 Server 端配合提供任务状态查询接口。5.3 连接生命周期长连接断开后的恢复策略MCP 会话本质上是长连接无论是 stdio 子进程还是 SSE 流Server 重启或网络波动都会让连接失效。这时候 LangGraph 的图本身还不知道工具调用会突然冒出连接类异常。我的做法是给 MCPManager 增加一个连接检测和重建机制。每次 call_tool 前先检测会话状态发现不可用就标记该 Server 为降级并把它的工具从当前工具列表里临时过滤掉同时把一条系统提示注入模型说明某个服务暂时不可用。当重建成功后再把工具列表恢复。这个降级流程避免了一次服务端故障导致整个 Agent 不可用的情况。5.4 调试建议先用纯客户端验证握手再接LangGraph最后再说一个我特别想分享的调试经验。不要一上来就把 MCP Server 接进 LangGraph先用简单的异步脚本直连 Server手动执行 initialize、tools/list、tools/call 三个步骤确认 Server 本身的行为符合预期。这一步能过滤掉大量其实是协议层问题的干扰。等直连验证通过后再把它包成 LangGraph 工具。我在实际项目里试过一次绕开这个过程直接接图结果区分图编排问题和协议问题花了一整天而先做直连验证之后同样的集成只花了不到两小时。依赖版本上我建议把 mcp 客户端 SDK 的版本锁死。这个项目迭代很快版本升级带来的协议行为变化不明显但排查起来非常费时间。拿我自己的体会来说MCP 多 Server 接入这种事协议复杂度远低于业务复杂度。你真正要设计的不是传输层而是工具命名空间、超时策略和连接降级这些工程性问题。先把这几件事想清楚再回头接 LangGraph会发现整个过程比预期顺利得多。