接手这个分享主题前我先说句实在的MCPModel Context Protocol模型上下文协议最近在圈里确实热得发烫但大部分教程都停留在“跑通一个 Server”的阶段真正到“多个 Server 同时接入、交给 LangGraph 统一编排”时坑就开始扎堆了。这篇文章我尽量不念文档直接从协议握手讲起一路写到 LangGraph 里多 Server 调用的完整落地最后把我在实操里踩过的坑和排查思路一并列出来适合已经跑过 MCP 基础 Demo、准备往生产级 Agent 方向走的开发者。1. MCP 是什么一次性看懂协议握手到底在做什么1.1 为什么大家都开始接 MCP从“工具适配器”到“统一 USB 口”我先打个比方。以前给 AI 接外部工具每个工具都是一套私有协议调飞书要写飞书 SDK查数据库要拼 SQL 连接搜网页要对接搜索 API。每接一个新工具就得给它单独写一套适配逻辑相当于给每种设备单独配充电线。MCP 做的事情就是把这些线统一成一根 Type-C 线工具方只要实现一个 MCP ServerAI 应用侧只要实现一个 MCP Client两边都按同一套 JSON-RPC 2.0 格式说话剩下的全部交给协议本身去协商。这套思路在工程上的意义非常直接你的 Agent 不再需要预知“我要调用哪几个工具、每个工具的参数签名是什么”而是在运行时通过tools/list动态发现工具再通过tools/call统一调用。工具升级了入参变了Agent 侧代码几乎不用动。这也是为什么很多人把 MCP 叫做“AI 原生时代的 API 标准化层”。1.2 initialize 握手的三个关键阶段MCP 的通信基于 JSON-RPC 2.0底层传输可以走 stdio标准输入输出也可以走 HTTP/SSE但不管哪种传输连接建立后的第一件事一定是“握手”。握手的完整过程有三个阶段。第一阶段是initialize请求。客户端发起连接后会先发送一个initialize请求里面带两样关键信息客户端自己声明的能力capabilities和希望使用的协议版本protocolVersion。服务端收到后会检查版本兼容性然后返回自己的capabilities。这里容易踩坑的地方是版本不兼容——服务端如果只支持一个旧版本而客户端声明了最新版本双方又没有做降级协商后续的tools/list都会被直接拒绝。第二阶段是notifications/initialized通知。这个阶段很容易被忽略客户端收到服务端的 initialize 响应后必须再单独发一条notifications/initialized通知告诉服务端“握手完成可以进入正常通信”。有些开发者只发完 initialize 就开始列工具时序上是不严谨的某些严格实现的 Server SDK 会直接忽略后续请求。第三阶段才是tools/list和tools/call。等initialized通知发出去之后客户端才能正式列举工具、读取资源、调用工具。整个握手的示意流程是这样的Client - Server: initialize (protocolVersion, capabilities) Server - Client: initialize result (serverInfo, capabilities) Client - Server: notifications/initialized Client - Server: tools/list Server - Client: tools list Client - Server: tools/call (name, arguments) Server - Client: tools/call result这块的逻辑顺序决定了多 Server 场景下的一个核心事实每个 Server 都需要独立完成上述握手不能共用一个连接。这也就是 LangGraph 里MultiServerMCPClient存在的意义——它对每个 Server 维护一套独立连接状态统一暴露成 LangChain 的 Tool 对象。提示如果你自己封装 MCP Client而不是用现成的 adapters一定要处理protocolVersion的协商逻辑。实际测试中服务端返回的版本号可能和你发过去的并不完全一致正确做法是以服务端返回的版本为准而不是强制要求完全相等。2. 多 Server 调用为什么难从环境冲突到路由决策2.1 真实场景一个 Agent 同时要查订单和搜资料单 Server 其实没什么可讲的——初始化一个连接、拉一次工具列表、丢给模型去调事情就结束了。真正的复杂度从“多个 Server”开始。我之前做过一个订单售后助手它需要同时访问两个数据源一个本地订单数据库SQL Server一个外部资料检索服务模拟的网页搜索。单看每个 Server实现都很简单但把它们放进同一个 Agent 的时候问题就来了。首先是工具命名冲突。两个 Server 都定义了一个叫query的工具数据库那边是query_orders搜索那边是query_web如果命名不规范LLM 在决定调用哪个工具时很容易混淆。其次是调用参数不统一数据库工具要求customer_id加days搜索工具要求keyword加max_results模型要在一个 Prompt 里同时理解两套参数语义对上下文设计和工具描述的要求一下就变高了。更麻烦的是状态管理。多个 Server 之间往往存在调用依赖——比如售后助手需要先从订单库里查出用户最近一笔订单再拿订单里的商品名称去搜索相关的售后政策。这种多步推理如果只靠模型自由发挥很容易在中途丢失中间状态。LangGraph 的价值恰好在这里它把“查询订单”和“搜索政策”拆成图里的节点把“订单信息”显式存进 state让每一步都有据可查、可回滚、可追踪。2.2 LangGraph 在解决什么问题状态、路由与循环LangGraph 的底层模型是状态图StateGraph核心要素有三个State全局状态、Node节点函数、Edge节点之间的连接。和 LangChain 以往的 Chain 不同LangGraph 允许你定义循环边、条件边所以天然适合写 Agent 的“思考 - 调用工具 - 再思考”循环。在多 Server 场景下LangGraph 给你的不是“同时调用多个工具”的魔法而是三个实在的好处。第一个好处是状态可见。订单查询的中间结果会存进 state后续任何节点都能读取不会因为多轮对话上下文太长而丢失关键信息。第二个好处是路由可控。你可以写一个专门的 router 节点根据 state 里的信息决定下一步走“继续调用工具”还是“直接给用户回复”而不是完全交给 LLM 的随机性。第三个好处是循环有边界。Agent 一旦陷入“工具调用死循环”比如连续调了十次搜索都没有得到理想结果你可以在条件边里设置最大迭代次数超过就强制结束。我用 LangGraph 做了几版对比之后确定的结论是多 Server 接入本身用MultiServerMCPClient就够了但它只解决“连接管理”问题真正让多个工具协作起来、形成稳定工作流的是 LangGraph 的状态图和路由逻辑。两者是配合关系不是替代关系。3. 实操LangGraph 多 MCP Server 调用完整实现3.1 环境准备与两个 Server 的定义先说环境。我这边用的是 Python 3.11核心依赖是这几个pip install langgraph langchain-mcp-adapters langchain-openai mcp再安装一下 SQL Server 连接需要的驱动和 SDK。顺便提醒一句如果机器上之前装过mcp旧版本建议直接升级到 1.x 以上老版本和langchain-mcp-adapters存在接口不兼容的问题。接下来定义两个 MCP Server。为了让你能直接跑我做了两个版本数据库版是真实连接 SQL Server搜索版我给了 mock 实现你把回调替换成自己的搜索接口即可。第一个文件server_orders.py订单查询工具走 SQL Serverfrom mcp.server.fastmcp import FastMCP import pyodbc mcp FastMCP(order_db) mcp.tool() def query_recent_orders(customer_id: str, days: int 30) - str: 查询指定客户最近 N 天的订单记录入参 customer_id 为客户编号days 为查询天数。 conn pyodbc.connect( DRIVER{ODBC Driver 17 for SQL Server}; SERVERlocalhost;DATABASEshop;UIDsa;PWDyour_password; TrustServerCertificateyes ) cursor conn.cursor() cursor.execute( SELECT order_id, order_date, product_name, amount FROM orders WHERE customer_id ? AND order_date DATEADD(day, -?, GETDATE()) , customer_id, days ) rows cursor.fetchall() return \n.join(str(row) for row in rows) if __name__ __main__: mcp.run()这里有两个细节要特别说明。第一查询条件里千万不要把日期直接拼成字符串传进去一定要用参数化查询否则不仅容易被注入还会踩到 SQL Server 的本地化日期格式坑——也就是热搜里经常出现的conversion failed when converting date and/or time from character string这个错误绝大多数情况都是因为传入的日期字符串格式和 SQL Server 会话的语言设置不一致。第二连接串里的TrustServerCertificateyes在本地测试时很重要否则 pyodbc 会因为证书校验失败而拒绝连接。第二个文件server_search.py资料检索工具我给了 mock 实现from mcp.server.fastmcp import FastMCP mcp FastMCP(web_search) DOCS { order_policy: 订单签收后 7 天内可申请无理由退货生鲜商品不支持。, refund_policy: 退款将在审核通过后 1-3 个工作日内原路退回。, delivery_policy: 默认快递为顺丰偏远地区加收 10 元配送费。, } mcp.tool() def search_policy(keyword: str, max_results: int 3) - str: 按关键词搜索售后政策文档。 matched [v for k, v in DOCS.items() if keyword in k or keyword in v] return \n.join(matched[:max_results]) if __name__ __main__: mcp.run()注意两个 Server 的入参设计是有意区分的一个用customer_id days一个用keyword max_results这样后面在 LangGraph 编排时才能看出多工具调用的参数路由效果。3.2 MultiServerMCPClient 初始化与工具过滤主程序里用langchain-mcp-adapters自带的MultiServerMCPClient来接两个 Server。这个 Client 的核心方法有两个一个是get_tools()返回经过包装的 LangChain Tool 列表另一个是上下文管理async with client确保所有 Server 连接在使用完后统一关闭。from langchain_mcp_adapters.client import MultiServerMCPClient client MultiServerMCPClient( { order_db: { command: python, args: [server_orders.py], transport: stdio, }, web_search: { command: python, args: [server_search.py], transport: stdio, }, } ) async def load_tools(): async with client: tools await client.get_tools() return tools这里有个我一开始没注意到、后来排查了很久才发现的坑get_tools()返回的工具名会带上 Server 前缀命名规则是 Server 名称加上原始工具名。比如 order_db 下的query_recent_orders会被改名为order_db_query_recent_orders。这个前缀在传给 LLM 时其实是有用的能帮模型区分工具来源但它也意味着你后续做工具过滤时要注意使用带前缀的名字否则直接写原始工具名会匹配不到。# 过滤示例只保留我关心的两个工具 tools await client.get_tools() filtered_tools [t for t in tools if t.name in { order_db_query_recent_orders, web_search_search_policy }]3.3 写一个可控的 LangGraph 编排有了工具之后最省事的方式是直接用create_react_agent一把梭。但我实际项目中更推荐自己搭一个 StateGraph理由前面说过可控、可追踪、能设置循环上限。下面是完整的主流程代码import json from typing import Annotated, TypedDict from langchain_openai import ChatOpenAI from langgraph.graph import StateGraph, START, END from langgraph.graph.message import add_messages from langgraph.prebuilt import ToolNode, tools_condition class AgentState(TypedDict): messages: Annotated[list, add_messages] max_steps: int def build_agent(): model ChatOpenAI(modelgpt-4o, temperature0) graph StateGraph(AgentState) graph.add_node(agent, lambda state: { messages: [model.bind_tools(ToolNode(tools)).invoke(state[messages])] }) graph.add_node(tools, ToolNode(tools)) graph.add_edge(START, agent) graph.add_conditional_edges(agent, tools_condition) graph.add_edge(tools, agent) return graph.compile()这段代码虽然短但有几个地方我特意做了工程化处理。第一是bind_tools把工具列表绑定到模型实例上这里的tools是第 3.2 节里过滤后的列表两个 Server 的工具都在里面模型只知道有这两个工具不知道背后是两个进程、两个 SDK、两套协议。第二是ToolNode来自langgraph.prebuilt它会自动解析模型输出的tool_calls然后把参数分发给正确的工具执行。第三是tools_condition这个内置的 router 节点用来判断模型是“需要调用工具”还是“可以直接回复”它相当于给图加了一条循环边agent 调用工具后结果会回到 agent让模型继续推理直到模型认为不需要再调用工具为止。如果对循环次数有硬性要求可以用自定义 router 替换tools_conditiondef should_continue(state: AgentState): last_message state[messages][-1] if last_message.tool_calls and state.get(max_steps, 0) 5: return tools return END这个 router 的意思很简单模型还想调用工具且步数没超上限就去执行工具否则结束。有了这个边界即使模型发疯连续调用几十次工具Agent 也会在第五次之后强制收敛。3.4 运行结果与调用链路分析启动 Agent 的入口用一个实际问题来验证“查一下客户 C10001 最近 7 天的订单然后根据第一笔订单的商品名去搜一下对应售后政策。”async def main(): response await agent.ainvoke({ messages: [{role: user, content: 查一下 C10001 最近 7 天的订单然后根据第一笔订单的商品名称搜索相关售后政策。}] }) for msg in response[messages]: if msg.type ai: print(msg.content) if __name__ __main__: import asyncio asyncio.run(main())如果你把messages完整打印出来会看到这样一条调用链路模型收到用户问题判断需要先查订单输出tool_calls参数为{customer_id: C10001, days: 7}。ToolNode调用order_db_query_recent_orders返回订单列表。订单结果作为附加消息传回模型模型提取出第一笔订单的商品名再次输出tool_calls调用web_search_search_policy。搜索工具返回售后政策模型综合订单信息和政策文本生成最终回答。这条链路证明了一件事MCP 的多 Server 接入不是让模型“同时思考两个工具”而是让模型“先想一步、执行一步、再想一步”。在 LangGraph 里每一步的状态都保存在 messages 列表里既不会丢失上下文也方便排查“到底是工具出了问题还是模型理解出了问题”。4. 常见问题与排查技巧实录4.1 连接类问题stdio 报错、目录错误、端口占用多 Server 模式下连接类问题我遇到的最多而且很多都特别隐蔽。典型的是 stdio 连接失败错误信息通常长这样unable to init server, could not connect或者干脆是进程直接退出。排查思路其实不复杂先把MultiServerMCPClient里的command从python改成python -u关闭缓冲同时把args里的脚本路径改成绝对路径。因为 stdio 是管道通信子进程的任何输出都会混进 JSON-RPC 消息流一旦子进程在启动时打印了无关日志握手就直接失败。我遇到过一个很奇葩的情况server_orders.py文件在 import pyodbc 的时候因为全局环境变量导致 SSL 库加载告警这个告警被打印到了 stdout结果 MCP Client 把告警内容当成了 JSON-RPC 消息去解析。排查了很久才发现解决方法是把 Server 端所有日志全部改成写 stderr保留 stdout 给协议数据。还有一类是 SSE/HTTP 传输模式下端口被占用表现为cannot start internal http server之类的报错。这个词条在 IDE 工具链里经常出现尤其是当你同时启动多个 Server 且默认端口相同的时候。不同 Server 一定要分配不同的端口MCP 不会帮你做端口自动协商。另外本地开发时如果之前有残留进程占用端口直接lsof -i :{port}或者netstat -ano | findstr {port}找出来杀掉比反复重启稳妥得多。4.2 工具与协议类问题工具找不到、参数格式错误第二类高频问题是tools/list阶段工具没加载出来或者调用时报参数错误。工具没加载出来多半是工具名被 Server 前缀改了而你在过滤时还用的原始名字。比如query_recent_orders实际上叫order_db_query_recent_orders。我踩过这个坑之后习惯做法是先无条件打印一遍所有工具名人工确认后再写过滤逻辑。参数格式错误则是另外一回事。MCP 的tools/call传参底层是严格 JSON 格式但 LangChain 的 Tool 会做一层转换。如果你在 Server 端定义工具时用了复杂的 Pydantic 模型作为入参转换过程中可能出现字段缺失或者类型不匹配。最典型的就是刚才提到的 SQL Server 日期时间转换错误——模型生成的日期字符串和 SQL Server 默认会话的日期格式不一致。解法是在 SQL 里用CONVERT(date, ?, 112)这类显式格式转换而不是让数据库隐式转换。这个看起来像是数据库问题但在 MCP 多 Server 链路里实际是“模型生成参数 - JSON 解析 - pyodbc 绑定变量”这个过程里产生的排查时要往前看不要只盯着数据库日志。4.3 调度与资源类问题死循环、上下文爆掉、并发困惑LangGraph 接入多 Server 之后资源类问题也开始变得明显。首先是死循环。模型在某一步可能反复调用同一个工具、拿回相同结果、再调用同一工具循环往复。用create_react_agent时这个问题很容易出现因为默认配置对迭代次数没有强约束。我自己的做法是升级到自定义 StateGraph加一个步数计数器放在 state 里超过阈值强制走END。这一步几乎是把 Agent 从“能跑”变成“稳定能用”的关键。其次是上下文爆掉。每个工具的返回结果都会拼进 messages如果数据库查询返回了大量行、搜索工具返回了大段文本多轮循环后 token 消耗会直线上升。我的处理方案有两条一是在 Server 端对返回内容做裁剪比如订单查询最多返回 10 条超出部分用统计信息替代明细二是在工具节点执行完后对结果做摘要只保留关键信息回灌给模型。这两条配合起来能把上下文长度控制在一个稳定的量级。最后是并发困惑。MultiServerMCPClient在设计上不是为“高频并发调用同一个 Server”准备的它的连接复用能力有限。如果同一个请求里模型要连续调用某个 Server 十几次性能会明显下降。更合理的做法是让 Server 端自己提供批量接口一次调用返回全部所需数据而不是让模型拆成几十次单点调用。这里不要过度相信模型会“聪明地合并请求”工具设计得不够粗粒度模型只会按最笨的方式反复调。5. 我的实操体会与一个小技巧最后分享一点自己的体会。MCP 协议本身并不复杂复杂的是把它嵌入到一个真实可用的 Agent 系统里。跑通单个 Server 只是起点真正考验工程能力的是多 Server 场景下的连接管理、管线编排和资源控制。LangGraph 在这里的角色不是简单的“调度器”而是给一套可能跑偏的工具调用逻辑划出边界、留下轨迹。一个很加分的小技巧是在调试多 Server 链路时把所有 MCP 消息都开启 debug 级别的日志输出。具体做法是在启动脚本前设置环境变量或直接在MultiServerMCPClient外层包一层消息记录器把initialize、tools/list、tools/call的原始 JSON-RPC 报文打出来。你会发现绝大多数“工具调用异常”在报文层就能看出原因根本不需要去翻模型提示词或者猜测工具内部逻辑。协议层的报错永远比业务层的猜测更准确。