1. 先说清楚MCP 到底解决什么问题LangGraph 为什么需要它MCP 这个协议最近在 AI 工程圈子里几乎成了标配尤其是当你开始用 LangGraph 编排多 Agent 时能不能把一个 MCP Server 接进来决定了你的工具系统是拼积木还是拆家。这篇东西是我实际把 MCP 协议握手到 LangGraph 多 Server 调用这条路完整跑通后留下的记录里面既有协议层面的拆解也有代码层面的实现更有一堆搜索和调试时踩过的坑。先回答一个最基础的问题MCP 是什么。模型上下文协议Model Context Protocol本质上是一套标准化的接口约定它的定位可以类比成 LLM 世界的 USB-C 接口。在它之前每个 Agent 接一个工具就要写一套自定义的集成逻辑接数据库是 HTTP API接浏览器是 Playwright 重封装接本地文件系统又得单独搞一套权限和路径处理。MCP 把工具、资源、提示词模板这三个核心能力统一成一套协议让任意客户端可以连接任意实现 MCP 规范的服务器相当于“一次握手处处可用”。LangGraph 需要 MCP 的原因也很直接。LangGraph 本身擅长编排状态机和多节点工作流但它并不内置天下所有工具的适配器。你用 LangGraph 写一个需求分析 Agent、一个代码生成 Agent、一个数据库查询 Agent每个 Agent 需要的工具集合完全不同。如果没有 MCP你得为每个 Agent 单独写工具注册、参数校验、调用失败的兜底逻辑有了 MCP你只需要让每个 Agent 去连对应的 Server工具列表、参数 schema、调用入口全部从 Server 侧拿到LangGraph 只管编排和执行。另一个容易忽略的点是MCP 的 Server 不一定是“LLM 官方提供的”。你完全可以在本地写一个 MCP Server包一层内部 API、内部数据库或自定义脚本甚至把旧系统的 Web Service 包成 MCP 工具。这套思路对做企业集成的人来说非常友好只需要在 Server 端实现协议客户端全行业通用不再为每家系统重复造轮子。对于想上手的朋友我的建议是先跑通一个最小闭环再考虑复杂场景。最小闭环包括一个 MCP Server哪怕只有一个加法工具、一个 MCP Client官方 SDK 或 LangChain 封装、一次成功的工具调用。这个闭环跑通后你理解握手、能力协商、工具列表和调用的底层逻辑都会变得非常具体。1.1 MCP 的本质是“接口约定”不是又一个大模型框架很多初学者容易把 MCP 理解成一个可以“启动起来运行 Agent”的框架这是一种方向性误解。MCP 不负责推理不负责管理对话状态也不负责编排 Agent它只负责解决“客户端要调用外部能力双方如何描述能力、如何请求、如何返回结果”这个问题。它给出了消息格式、传输方式和会话生命周期所有这些设计都是为了让工具系统可以被动态接入、动态发现而不是写死在代码里。打个不严谨但利于理解的比方如果 Agent 是大脑那么 MCP 是中枢神经系统的信号标准Server 是各个器官。大脑不需要知道胃怎么消化食物胃也不需要知道大脑的推理逻辑双方只要按同一套神经信号标准传数据就能工作。MCP 就是这套信号标准它让“大脑”和“器官”之间的耦合降到了最低。1.2 在 LangGraph 的场景里多 Server 调用到底指什么“多 Server 调用”这个说法听起来很高端落地到 LangGraph 里其实包含三个层次。第一层是连接层一个进程中同时与多个 MCP Server 建立会话每个 Server 可能是 stdio 子进程也可能是一个远程 HTTP 服务。第二层是工具汇聚层把多个 Server 暴露的工具合并成一个统一的工具列表交给 LLM由模型在推理过程中自行决定调用哪个。第三层是执行调度层LangGraph 节点收到 LLM 的工具调用请求后要能准确路由到对应 Server执行并拿到结果返回给模型或写入状态。这三个层次每一层都有坑。连接层最常见的问题是传输方式不匹配有 Server 只支持 stdio你的客户端却配置成了 streamable HTTP很快就出现握手失败。工具汇聚层最常见的问题是两个 Server 暴露了同名工具比如文件搜索工具都叫“search_files”模型要调用时根本无法区分。执行调度层的坑更多比如某个 Server 自身的会话状态维护在服务端你的图并行执行多个分支时就会收到意外的状态污染。我后面会针对每一层给出具体的写法和避坑建议。2. 协议握手是入门的第一道门槛initialize 到底在忙什么MCP 的会话建立不是 TCP 三次握手那么“裸”它是基于 JSON-RPC 2.0 的一套应用层协议。整个连接过程最核心的环节是 initialize 请求这一段走不明白后面所有工具调用都会滑稽地失败。2.1 握手过程拆解从 initialize 到 initializedMCP 的握手大致可以分成四步。第一步客户端向服务端发送 initialize 请求请求里必须带 protocolVersion、capabilities、clientInfo 三个字段。第二步服务端返回协议版本、服务端能力和 serverInfo。第三步客户端发送 notifications/initialized 通知表示“我知道了你的能力我们可以正式开始工作”。第四步双方进入正常工作阶段此时客户端才能发送 tools/list、tools/call、resources/list 等请求。我贴一段实际的请求和响应体这对理解协议非常重要。客户端发的 initialize 请求长这样{ jsonrpc: 2.0, id: 1, method: initialize, params: { protocolVersion: 2025-03-26, capabilities: { roots: { listChanged: true } }, clientInfo: { name: langgraph-mcp-client, version: 0.1.0 } } }服务端收到后会回一个响应示例{ jsonrpc: 2.0, id: 1, result: { protocolVersion: 2025-03-26, capabilities: { tools: { listChanged: true }, resources: {} }, serverInfo: { name: hello-mcp-server, version: 0.1.0 } } }这段看起来简单但协议版本这块有个很现实的困惑服务端返回的 protocolVersion 未必等于客户端发出的版本。如果客户端发送的是新版本 “2025-03-26”而服务端只实现了更早的 “2024-11-05”服务端会返回自己支持的版本。此时客户端有两种选择接受降级或者拒绝连接。绝大多数情况下我们应该接受降级因为工具调用基本不受小版本差异影响。我见过一些不成熟客户端在 initialize 响应里发现版本不等就直接抛异常导致明明 Server 可用却白白断掉连接。2.2 能力协商不是可有可无capabilities 的含义很多实现代码会为了省事直接发空的 capabilities这种写法短期内能跑通长期一定出问题。capabilities 的作用是告诉对方“我支持哪些主动通知、哪些扩展能力”它决定后续双方能不能使用 listChanged 这类推送机制。具体来说客户端侧通常会声明 roots意思是“我允许服务端读取当前工作区的根目录集合”。服务端侧通常会声明 tools、resources、prompts 三类能力中它实现了哪些。如果你的 Server 只实现了 tools那么客户端就没必要去请求 resources/list可以直接跳过相关逻辑节省一次无意义的往返。在 LangGraph 场景下我建议客户端侧一定要认真填写 clientInfo因为你在调试多 Server 时服务端日志会显示“当前连接来自哪个客户端”这个信息能帮你快速定位到底是哪个图、哪个节点建立的连接。2.3 传输方式差异会导致握手行为不同stdio 与 streamable HTTPMCP 规定了两种主流传输方式它们对握手的影响差异巨大。stdio 是客户端拉起一个子进程通过 stdin 写消息、从 stdout 读消息单条 JSON-RPC 消息以换行符分隔。它的优点是无需网络配置适合本地工具缺点是生命周期完全依赖父进程子进程崩了你只能从 stderr 里看日志。streamable HTTP 则是通过 HTTP POST/GET 发送请求支持 SSE 服务端推送。它的优点是可以远程部署、多客户端复用缺点是要考虑鉴权、超时、跨域以及服务端是否支持 GET 方式创建长期会话。我在实际项目中遇到过一种情况同一个 Server 用 stdio 连接完全正常但用 streamable HTTP 就一直在 initialize 阶段超时。排查后发现是服务端部署在反代后面GET 请求被拦截。如果你用 Docker 部署 MCP Server务必确认暴露了正确的 HTTP 端口并且客户端配置了正确的 base URL。以后排查握手问题时第一步永远先确认“传输方式是否匹配”别急着怀疑协议实现。3. 先写一个最小 MCP Server把链路跑通理论讲再多不如写一个最小实现有用。下面我用 Python 官方 SDK 的 FastMCP 来写一个极简 Server然后跑一遍完整链路。选 Python 是因为生态最成熟调试起来最省心。3.1 环境准备与项目结构首先准备 Python 3.10 以上环境创建一个虚拟目录mkdir mcp-demo cd mcp-demo python -m venv .venv source .venv/bin/activate pip install mcp[cli] langgraph langchain-openai这里我把 LangGraph 和 OpenAI 客户端一起装上后面第五节的调用会直接复用这个环境。项目结构不需要复杂建议拆成三个文件hello_server.pyMCP Server只暴露一个加法工具和一个文件资源client.py验证用客户端独立跑通握手与工具调用langgraph_app.py接入 LangGraph 多 Server 调用的主程序3.2 用 FastMCP 暴露一个工具FastMCP 的写法非常接近“函数装饰器”风格几乎零学习成本。hello_server.py 内容如下from mcp.server.fastmcp import FastMCP mcp FastMCP(hello-server) mcp.tool() def add_numbers(a: int, b: int) - int: Add two integers together. return a b if __name__ __main__: mcp.run()这段代码已经把 add_numbers 暴露成 MCP 工具工具名就是 add_numbers描述来自 docstring输入参数 schema 由 FastMCP 根据类型注解自动生成。这个自动生成机制非常方便但有个经验之谈凡是工具参数里有复杂嵌套结构建议写 Pydantic 模型显式声明否则描述性字段和必填约束可能不符合预期。启动 Server 有两种方式开发期建议用 MCP Inspector 或者 mcp dev。直接在终端跑python hello_server.py会看到进程在 stdio 模式的等待状态此时没有外部客户端它不会输出任何东西这是正常现象。3.3 用官方客户端把握手和工具调用跑一遍client.py 的核心逻辑是创建 stdio 子进程套上 ClientSession然后先 initialize再循环调用工具。代码示例如下import asyncio from mcp import ClientSession, StdioServerParameters from mcp.client.stdio import stdio_client async def main(): server_params StdioServerParameters( commandpython, args[hello_server.py] ) async with stdio_client(server_params) as (read, write): async with ClientSession(read, write) as session: # 这一步完成握手 init_result await session.initialize() print(Server info:, init_result.serverInfo) # 列出工具 tools await session.list_tools() for tool in tools.tools: print(Tool:, tool.name, tool.inputSchema) # 调用工具 result await session.call_tool(add_numbers, {a: 1, b: 2}) print(Result:, result) asyncio.run(main())运行后你会依次看到握手数据、工具列表和一个包含 text 块的结果。整个流程非常直观。很多新人卡在这里最常见的问题是 Windows 下子进程里的 Python 路径不对导致 stdio_client 拉起子进程失败。建议把 command 写成 sys.executable确保指向当前虚拟环境的解释器。server_params StdioServerParameters( commandsys.executable, args[hello_server.py] )这条小改动可以帮你省掉至少一小时的 Windows 环境排查。4. LangGraph 多 Server 调用架构与实现最小链路跑通后下一步就是把它搬到 LangGraph 里。很多人一上来就想写很复杂的多 Agent 编排但我的建议是从一个最简单的“单节点调用多个 Server”开始先验证工具聚合没有问题再逐步做路由拆分。4.1 思路一个 Client 一个 Server工具汇流给 AgentLangGraph 本身不内置 MCP 适配器它靠的是 LangChain 的工具接口。LangChain 的 BaseTool 有 name、description、args_schema 等字段我们要做的工作就是把 MCP Server 的 Tool 转换成 LangChain 的 StructuredTool或者直接用 langchain-mcp-adapters 这个官方适配包。我推荐直接用 langchain-mcp-adapters它已经封装好了从 MCP Client 到 LangChain 工具的转换。基本思路是每个 Server 对应一个 ClientSessionsession 初始化完成后调用 list_tools再调用 load_mcp_tools 得到可被 LLM 调用的工具列表。多个 Server 就是重复这个过程最后把所有返回的工具列表合并一起 bind 给模型。这里有一个架构上的取舍是用一个图节点连接全部 Server还是一个 Server 对应一个子图节点。我倾向于前者作为起点因为工具数量少单节点绑定所有工具可以降低图调试的复杂度。如果工具数量很大超过二十个再考虑按域拆子图每个子图只暴露部分工具然后由上层路由节点决定进入哪个子图。4.2 工具聚合与命名空间避免“同名工具打架”多 Server 聚合最现实的问题就是工具重名。我做过一个项目里有文件搜索 Server 和 Web 搜索 Server两边都叫 search直接把两个工具放进同一列表后LLM 每次调 search 都行为不定因为 langchain 工具名默认取自 MCP 工具名重名会导致覆盖。解决办法是给每个 Server 的工具名加前缀。langchain-mcp-adapters 的 load_mcp_tools 支持一个 tool_name 定制参数如果你读源码它会为每个工具创建一个 Tool名称可以基于原始工具名映射。最简单的落地方式是先拿到原始 MCP 工具列表再自己构造 StructuredToolfrom langchain_core.tools import StructuredTool from langchain_core.utils.function_calling import convert_to_openai_function def convert_mcp_tool_to_langchain(tool, server_prefix): async def run_tool(**kwargs): result await session.call_tool(tool.name, kwargs) return .join(item.text for item in result.content if item.type text) return StructuredTool( namef{server_prefix}_{tool.name}, descriptiontool.description or (no description), args_schemajson_type_to_pydantic(tool.inputSchema), coroutinerun_tool )命名规则我偏爱用 server 名加下划线作为前缀比如 db_query、web_search。这样 LLM 看到的工具名是 db_query_users 和 web_search_keyword模型很少会搞混且日志排查时一看名字就知道是哪家的工具。注意工具描述一定要写清楚“应该何时使用这个工具、尽量不要何时使用”这种描述远比漂亮的参数 schema 重要。MCP 工具本身包含 description但如果 Server 作者写得敷衍你聚合后模型就很容易误调。我在生产里通常会在转换层再包一层描述把业务语义补全。4.3 在 Graph 节点里做路由与执行拆分单节点聚合全部工具虽然简单但执行效率偏低因为 LLM 每轮要面对十几二十个工具进行选择。如果工具数量上升我强烈建议拆两步一个路由节点先判断“当前问题该找哪个 Server”然后分派到对应子节点执行。LangGraph 的 StateGraph 本身就适合这件事。比如定义 state 里有一个 intent 字段路由节点根据用户输入映射到 target_server然后条件边把流程切到 search_node 或 db_node。每个子节点内部再绑定自己的工具列表这样模型每轮的候选工具只有 3-5 个准确率和响应速度都更好。这种做法的开销是路由节点需要额外一次 LLM 调用但对于复杂 Agent 场景这点开销换来的稳定性和可维护性完全值得。5. 实操过程与核心环节实现本章给出一个可以直接抄作业的 LangGraph 多 Server 调用实现包含两个虚拟 Server一个提供文件查询工具一个提供时间工具。为了控制篇幅我只贴核心片段完整逻辑会写清楚。5.1 多 Server 初始化顺序与搬运工具先准备好两个 Server 的启动参数然后创建一个context管理函数来同时初始化多个会话import asyncio from contextlib import asynccontextmanager from mcp import ClientSession, StdioServerParameters from mcp.client.stdio import stdio_client SERVERS { file: StdioServerParameters(commandsys.executable, args[file_server.py]), time: StdioServerParameters(commandsys.executable, args[time_server.py]), } asynccontextmanager async def create_mcp_clients(): sessions {} for name, params in SERVERS.items(): read, write await stdio_client(params).__aenter__() session await ClientSession(read, write).__aenter__() await session.initialize() sessions[name] session try: yield sessions finally: for name, session in sessions.items(): await session.__aexit__(None, None, None)这个初始化顺序有讲究一定要先 initialize再 list_tools否则部分 Server 可能因为状态未就绪返回空工具列表。我见过不少代码在创建 session 后立刻 list_tools结果偶尔拿到空列表后来才发现是initialize 和 list_tools 并发执行导致时序竞争。严格串行初始化并把 session 存入字典是最笨但最可靠的方式。5.2 执行工具调用并处理结果LangGraph 节点内部LLM 返回一个 tool_calls 数组每个元素包含 name、args、id。我们需要根据工具名前缀找到对应 Server然后调用 call_tool 方法。下面是核心执行函数def make_agent_node(sessions, llm): tools [] for server_name, session in sessions.items(): mcp_tools await session.list_tools() for mcp_tool in mcp_tools.tools: tools.append(convert_mcp_tool_to_langchain(mcp_tool, server_name, session)) llm_with_tools llm.bind_tools(tools) async def agent_node(state): messages state[messages] response await llm_with_tools.ainvoke(messages) return {messages: [response]} return agent_node然后执行节点逐个处理工具调用async def tool_node(state, sessions): last_message state[messages][-1] outputs [] if not hasattr(last_message, tool_calls): return {messages: state[messages]} for tool_call in last_message.tool_calls: tool_name tool_call[name] server tool_name.split(_, 1)[0] session sessions[server] result await session.call_tool( tool_name.split(_, 1)[1], tool_call[args] ) text_parts [item.text for item in result.content if item.type text] outputs.append(ToolMessage(content\n.join(text_parts), tool_call_idtool_call[id])) return {messages: outputs}这段代码里两个细节容易踩坑。一个是 ToolMessage 的 tool_call_id 必须和原始 tool_call 的 id 一致否则模型无法关联工具结果。另一个是 call_tool 返回的 content 可能包含 image、audio 等非文本块聚合文本时必须过滤 item.type text否则会拼出大量空格和难懂的对象字符串。5.3 流式输出与并发LangGraph 天然支持状态流式但它默认不流式输出 MCP 工具的中间结果因为工具调用通常是一次性返回。如果你想让 Server 的进度通过 SSE 或 stdio 流式传出需要在 Server 端实现 notifications 或者用长耗时任务的进度通知机制。目前官方 SDK 对工具执行过程内的流式支持还在演进常规做法依然是工具快速返回最终结果LLM 再流式生成最终答案。并发方面最需要注意的是避免多个节点共享同一个 ClientSession 并发调用call_tool。MCP 规范并不强制 Server 处理并发请求Stdio 子进程往往不处理并行调用轻则排队重则直接崩。我的建议是在每个节点内部加上信号量控制并发上限或者干脆保持一个 Session 同一时刻只处理一个 call_toolsemaphore asyncio.Semaphore(1) async def safe_call_tool(session, name, args): async with semaphore: return await session.call_tool(name, args)这个简单限制能避免 90% 的随机性错误。6. 踩坑记录协议版本、超时、工具 schema 校验这一章是真正值钱的部分。我在把 MCP 嵌入 LangGraph 的过程中踩过一堆坑整理成速查表按频率排序。6.1 transport 不匹配 / 服务器退出现象客户端 initialize 阶段直接超时甚至报 “Server closed the connection without sending a response”。排查第一步确认 Server 是否真的在运行运行方式是否与启动参数匹配。如果你 command 写的是 pythonargs 写的是 server.py但当前环境里能访问的 python 是系统解释器而不是虚拟环境解释器就可能因为依赖缺失导致子进程秒退。第二步检查 stdio 是否被污染。这是个大坑如果你的 Server 代码里有任何 print 输出到 stdoutstdout 就不再是纯 JSON-RPC 消息通道客户端解析第一条消息就会失败。常见病源是调试用的 print 忘删或者第三方库往 stdout 打了日志。规范做法是所有调试信息写 stderrprint(debug log, filesys.stderr)另外Windows 上创建子进程时某些环境变量会干扰 stdio建议在 StdioServerParameters 里显式带一个干净的 env最少保留 PATH、PYTHONPATH 和系统盘符相关变量。6.2 工具调用被拒或报错现象工具能列出但 call_tool 返回 isErrorTrue。常见原因有三个参数类型不匹配、必填参数缺省、Server 内部异常。MCP 的 inputSchema 是 JSON Schema客户端传的参数要严格符合它。LangChain 的 StructuredTool 会自动做 Pydantic 校验但如果你手工构造工具函数很可能没做参数强制转换。遇到这类问题最有效的手段是给 call_tool 外层包一个 try/except把异常信息连同工具名一起返回给模型。很多情况下模型看到详细报错能自我纠错再次调用。我在生产项目里就是靠这个“错误回传重试”机制把一次调用成功率从 70% 提到 98%。6.3 服务器连接生命周期与状态污染现象图跑多轮后工具行为不一致或者状态残留。如果 Server 内部维护了会话状态比如“当前选择的数据库”那么多个不同图分支共用同一个 Server 会话就会互相污染。要避免这类问题要么把 Server 设计成无状态——每次 call_tool 都携带完整上下文要么为每个图实例创建独立的 ClientSession。我的原则是MCP Server 尽量做成纯函数式所有上下文通过参数传入不要靠服务端缓存。这样多 Server 调用才能保持可预期性。6.4 一个易被忽略的问题工具 schema 复杂度过高有些 Server 暴露的工具 inputSchema 极其复杂嵌套五六层。LLM 构造参数时很容易遗漏深层字段导致校验失败。遇到这种情况光靠模型自己猜参数不可靠。建议在工具描述里写明“参数默认值是什么、哪些字段可以省略”甚至给出一两个完整的 JSON 示例。这个描述内容会进入 prompt是调整模型调用成功率最直接的手段远比你改代码里的校验逻辑快。调试 MCP 还有一个神器建议使用MCP Inspector。它是一个图形化工具可以直接查看工具列表、修改参数调用工具、查看原始 JSON-RPC 消息。遇到诡异问题先用 Inspector 复现再回到代码里排查至少能节省一半时间。官方还有 mcp dev 可以快速启动 Server 并附加 Inspector开发体验很不错。最后多 Server 调用的稳定性归根结底取决于你的容错设计而不是协议本身。给每个 Server 调用设置超时给每次失败备好 fallback在 LangGraph 里预留一个“工具调用失败但允许模型继续”的条件分支这套思路比纠结协议版本更新更有价值。我在实际项目中就是靠这种防御式写法才敢把四个 MCP Server 同时接进一条 Agent 工作流里稳定跑数天。
MCP×LangGraph多Server调用全解：从协议原理到工程落地
1. 先说清楚MCP 到底解决什么问题LangGraph 为什么需要它MCP 这个协议最近在 AI 工程圈子里几乎成了标配尤其是当你开始用 LangGraph 编排多 Agent 时能不能把一个 MCP Server 接进来决定了你的工具系统是拼积木还是拆家。这篇东西是我实际把 MCP 协议握手到 LangGraph 多 Server 调用这条路完整跑通后留下的记录里面既有协议层面的拆解也有代码层面的实现更有一堆搜索和调试时踩过的坑。先回答一个最基础的问题MCP 是什么。模型上下文协议Model Context Protocol本质上是一套标准化的接口约定它的定位可以类比成 LLM 世界的 USB-C 接口。在它之前每个 Agent 接一个工具就要写一套自定义的集成逻辑接数据库是 HTTP API接浏览器是 Playwright 重封装接本地文件系统又得单独搞一套权限和路径处理。MCP 把工具、资源、提示词模板这三个核心能力统一成一套协议让任意客户端可以连接任意实现 MCP 规范的服务器相当于“一次握手处处可用”。LangGraph 需要 MCP 的原因也很直接。LangGraph 本身擅长编排状态机和多节点工作流但它并不内置天下所有工具的适配器。你用 LangGraph 写一个需求分析 Agent、一个代码生成 Agent、一个数据库查询 Agent每个 Agent 需要的工具集合完全不同。如果没有 MCP你得为每个 Agent 单独写工具注册、参数校验、调用失败的兜底逻辑有了 MCP你只需要让每个 Agent 去连对应的 Server工具列表、参数 schema、调用入口全部从 Server 侧拿到LangGraph 只管编排和执行。另一个容易忽略的点是MCP 的 Server 不一定是“LLM 官方提供的”。你完全可以在本地写一个 MCP Server包一层内部 API、内部数据库或自定义脚本甚至把旧系统的 Web Service 包成 MCP 工具。这套思路对做企业集成的人来说非常友好只需要在 Server 端实现协议客户端全行业通用不再为每家系统重复造轮子。对于想上手的朋友我的建议是先跑通一个最小闭环再考虑复杂场景。最小闭环包括一个 MCP Server哪怕只有一个加法工具、一个 MCP Client官方 SDK 或 LangChain 封装、一次成功的工具调用。这个闭环跑通后你理解握手、能力协商、工具列表和调用的底层逻辑都会变得非常具体。1.1 MCP 的本质是“接口约定”不是又一个大模型框架很多初学者容易把 MCP 理解成一个可以“启动起来运行 Agent”的框架这是一种方向性误解。MCP 不负责推理不负责管理对话状态也不负责编排 Agent它只负责解决“客户端要调用外部能力双方如何描述能力、如何请求、如何返回结果”这个问题。它给出了消息格式、传输方式和会话生命周期所有这些设计都是为了让工具系统可以被动态接入、动态发现而不是写死在代码里。打个不严谨但利于理解的比方如果 Agent 是大脑那么 MCP 是中枢神经系统的信号标准Server 是各个器官。大脑不需要知道胃怎么消化食物胃也不需要知道大脑的推理逻辑双方只要按同一套神经信号标准传数据就能工作。MCP 就是这套信号标准它让“大脑”和“器官”之间的耦合降到了最低。1.2 在 LangGraph 的场景里多 Server 调用到底指什么“多 Server 调用”这个说法听起来很高端落地到 LangGraph 里其实包含三个层次。第一层是连接层一个进程中同时与多个 MCP Server 建立会话每个 Server 可能是 stdio 子进程也可能是一个远程 HTTP 服务。第二层是工具汇聚层把多个 Server 暴露的工具合并成一个统一的工具列表交给 LLM由模型在推理过程中自行决定调用哪个。第三层是执行调度层LangGraph 节点收到 LLM 的工具调用请求后要能准确路由到对应 Server执行并拿到结果返回给模型或写入状态。这三个层次每一层都有坑。连接层最常见的问题是传输方式不匹配有 Server 只支持 stdio你的客户端却配置成了 streamable HTTP很快就出现握手失败。工具汇聚层最常见的问题是两个 Server 暴露了同名工具比如文件搜索工具都叫“search_files”模型要调用时根本无法区分。执行调度层的坑更多比如某个 Server 自身的会话状态维护在服务端你的图并行执行多个分支时就会收到意外的状态污染。我后面会针对每一层给出具体的写法和避坑建议。2. 协议握手是入门的第一道门槛initialize 到底在忙什么MCP 的会话建立不是 TCP 三次握手那么“裸”它是基于 JSON-RPC 2.0 的一套应用层协议。整个连接过程最核心的环节是 initialize 请求这一段走不明白后面所有工具调用都会滑稽地失败。2.1 握手过程拆解从 initialize 到 initializedMCP 的握手大致可以分成四步。第一步客户端向服务端发送 initialize 请求请求里必须带 protocolVersion、capabilities、clientInfo 三个字段。第二步服务端返回协议版本、服务端能力和 serverInfo。第三步客户端发送 notifications/initialized 通知表示“我知道了你的能力我们可以正式开始工作”。第四步双方进入正常工作阶段此时客户端才能发送 tools/list、tools/call、resources/list 等请求。我贴一段实际的请求和响应体这对理解协议非常重要。客户端发的 initialize 请求长这样{ jsonrpc: 2.0, id: 1, method: initialize, params: { protocolVersion: 2025-03-26, capabilities: { roots: { listChanged: true } }, clientInfo: { name: langgraph-mcp-client, version: 0.1.0 } } }服务端收到后会回一个响应示例{ jsonrpc: 2.0, id: 1, result: { protocolVersion: 2025-03-26, capabilities: { tools: { listChanged: true }, resources: {} }, serverInfo: { name: hello-mcp-server, version: 0.1.0 } } }这段看起来简单但协议版本这块有个很现实的困惑服务端返回的 protocolVersion 未必等于客户端发出的版本。如果客户端发送的是新版本 “2025-03-26”而服务端只实现了更早的 “2024-11-05”服务端会返回自己支持的版本。此时客户端有两种选择接受降级或者拒绝连接。绝大多数情况下我们应该接受降级因为工具调用基本不受小版本差异影响。我见过一些不成熟客户端在 initialize 响应里发现版本不等就直接抛异常导致明明 Server 可用却白白断掉连接。2.2 能力协商不是可有可无capabilities 的含义很多实现代码会为了省事直接发空的 capabilities这种写法短期内能跑通长期一定出问题。capabilities 的作用是告诉对方“我支持哪些主动通知、哪些扩展能力”它决定后续双方能不能使用 listChanged 这类推送机制。具体来说客户端侧通常会声明 roots意思是“我允许服务端读取当前工作区的根目录集合”。服务端侧通常会声明 tools、resources、prompts 三类能力中它实现了哪些。如果你的 Server 只实现了 tools那么客户端就没必要去请求 resources/list可以直接跳过相关逻辑节省一次无意义的往返。在 LangGraph 场景下我建议客户端侧一定要认真填写 clientInfo因为你在调试多 Server 时服务端日志会显示“当前连接来自哪个客户端”这个信息能帮你快速定位到底是哪个图、哪个节点建立的连接。2.3 传输方式差异会导致握手行为不同stdio 与 streamable HTTPMCP 规定了两种主流传输方式它们对握手的影响差异巨大。stdio 是客户端拉起一个子进程通过 stdin 写消息、从 stdout 读消息单条 JSON-RPC 消息以换行符分隔。它的优点是无需网络配置适合本地工具缺点是生命周期完全依赖父进程子进程崩了你只能从 stderr 里看日志。streamable HTTP 则是通过 HTTP POST/GET 发送请求支持 SSE 服务端推送。它的优点是可以远程部署、多客户端复用缺点是要考虑鉴权、超时、跨域以及服务端是否支持 GET 方式创建长期会话。我在实际项目中遇到过一种情况同一个 Server 用 stdio 连接完全正常但用 streamable HTTP 就一直在 initialize 阶段超时。排查后发现是服务端部署在反代后面GET 请求被拦截。如果你用 Docker 部署 MCP Server务必确认暴露了正确的 HTTP 端口并且客户端配置了正确的 base URL。以后排查握手问题时第一步永远先确认“传输方式是否匹配”别急着怀疑协议实现。3. 先写一个最小 MCP Server把链路跑通理论讲再多不如写一个最小实现有用。下面我用 Python 官方 SDK 的 FastMCP 来写一个极简 Server然后跑一遍完整链路。选 Python 是因为生态最成熟调试起来最省心。3.1 环境准备与项目结构首先准备 Python 3.10 以上环境创建一个虚拟目录mkdir mcp-demo cd mcp-demo python -m venv .venv source .venv/bin/activate pip install mcp[cli] langgraph langchain-openai这里我把 LangGraph 和 OpenAI 客户端一起装上后面第五节的调用会直接复用这个环境。项目结构不需要复杂建议拆成三个文件hello_server.pyMCP Server只暴露一个加法工具和一个文件资源client.py验证用客户端独立跑通握手与工具调用langgraph_app.py接入 LangGraph 多 Server 调用的主程序3.2 用 FastMCP 暴露一个工具FastMCP 的写法非常接近“函数装饰器”风格几乎零学习成本。hello_server.py 内容如下from mcp.server.fastmcp import FastMCP mcp FastMCP(hello-server) mcp.tool() def add_numbers(a: int, b: int) - int: Add two integers together. return a b if __name__ __main__: mcp.run()这段代码已经把 add_numbers 暴露成 MCP 工具工具名就是 add_numbers描述来自 docstring输入参数 schema 由 FastMCP 根据类型注解自动生成。这个自动生成机制非常方便但有个经验之谈凡是工具参数里有复杂嵌套结构建议写 Pydantic 模型显式声明否则描述性字段和必填约束可能不符合预期。启动 Server 有两种方式开发期建议用 MCP Inspector 或者 mcp dev。直接在终端跑python hello_server.py会看到进程在 stdio 模式的等待状态此时没有外部客户端它不会输出任何东西这是正常现象。3.3 用官方客户端把握手和工具调用跑一遍client.py 的核心逻辑是创建 stdio 子进程套上 ClientSession然后先 initialize再循环调用工具。代码示例如下import asyncio from mcp import ClientSession, StdioServerParameters from mcp.client.stdio import stdio_client async def main(): server_params StdioServerParameters( commandpython, args[hello_server.py] ) async with stdio_client(server_params) as (read, write): async with ClientSession(read, write) as session: # 这一步完成握手 init_result await session.initialize() print(Server info:, init_result.serverInfo) # 列出工具 tools await session.list_tools() for tool in tools.tools: print(Tool:, tool.name, tool.inputSchema) # 调用工具 result await session.call_tool(add_numbers, {a: 1, b: 2}) print(Result:, result) asyncio.run(main())运行后你会依次看到握手数据、工具列表和一个包含 text 块的结果。整个流程非常直观。很多新人卡在这里最常见的问题是 Windows 下子进程里的 Python 路径不对导致 stdio_client 拉起子进程失败。建议把 command 写成 sys.executable确保指向当前虚拟环境的解释器。server_params StdioServerParameters( commandsys.executable, args[hello_server.py] )这条小改动可以帮你省掉至少一小时的 Windows 环境排查。4. LangGraph 多 Server 调用架构与实现最小链路跑通后下一步就是把它搬到 LangGraph 里。很多人一上来就想写很复杂的多 Agent 编排但我的建议是从一个最简单的“单节点调用多个 Server”开始先验证工具聚合没有问题再逐步做路由拆分。4.1 思路一个 Client 一个 Server工具汇流给 AgentLangGraph 本身不内置 MCP 适配器它靠的是 LangChain 的工具接口。LangChain 的 BaseTool 有 name、description、args_schema 等字段我们要做的工作就是把 MCP Server 的 Tool 转换成 LangChain 的 StructuredTool或者直接用 langchain-mcp-adapters 这个官方适配包。我推荐直接用 langchain-mcp-adapters它已经封装好了从 MCP Client 到 LangChain 工具的转换。基本思路是每个 Server 对应一个 ClientSessionsession 初始化完成后调用 list_tools再调用 load_mcp_tools 得到可被 LLM 调用的工具列表。多个 Server 就是重复这个过程最后把所有返回的工具列表合并一起 bind 给模型。这里有一个架构上的取舍是用一个图节点连接全部 Server还是一个 Server 对应一个子图节点。我倾向于前者作为起点因为工具数量少单节点绑定所有工具可以降低图调试的复杂度。如果工具数量很大超过二十个再考虑按域拆子图每个子图只暴露部分工具然后由上层路由节点决定进入哪个子图。4.2 工具聚合与命名空间避免“同名工具打架”多 Server 聚合最现实的问题就是工具重名。我做过一个项目里有文件搜索 Server 和 Web 搜索 Server两边都叫 search直接把两个工具放进同一列表后LLM 每次调 search 都行为不定因为 langchain 工具名默认取自 MCP 工具名重名会导致覆盖。解决办法是给每个 Server 的工具名加前缀。langchain-mcp-adapters 的 load_mcp_tools 支持一个 tool_name 定制参数如果你读源码它会为每个工具创建一个 Tool名称可以基于原始工具名映射。最简单的落地方式是先拿到原始 MCP 工具列表再自己构造 StructuredToolfrom langchain_core.tools import StructuredTool from langchain_core.utils.function_calling import convert_to_openai_function def convert_mcp_tool_to_langchain(tool, server_prefix): async def run_tool(**kwargs): result await session.call_tool(tool.name, kwargs) return .join(item.text for item in result.content if item.type text) return StructuredTool( namef{server_prefix}_{tool.name}, descriptiontool.description or (no description), args_schemajson_type_to_pydantic(tool.inputSchema), coroutinerun_tool )命名规则我偏爱用 server 名加下划线作为前缀比如 db_query、web_search。这样 LLM 看到的工具名是 db_query_users 和 web_search_keyword模型很少会搞混且日志排查时一看名字就知道是哪家的工具。注意工具描述一定要写清楚“应该何时使用这个工具、尽量不要何时使用”这种描述远比漂亮的参数 schema 重要。MCP 工具本身包含 description但如果 Server 作者写得敷衍你聚合后模型就很容易误调。我在生产里通常会在转换层再包一层描述把业务语义补全。4.3 在 Graph 节点里做路由与执行拆分单节点聚合全部工具虽然简单但执行效率偏低因为 LLM 每轮要面对十几二十个工具进行选择。如果工具数量上升我强烈建议拆两步一个路由节点先判断“当前问题该找哪个 Server”然后分派到对应子节点执行。LangGraph 的 StateGraph 本身就适合这件事。比如定义 state 里有一个 intent 字段路由节点根据用户输入映射到 target_server然后条件边把流程切到 search_node 或 db_node。每个子节点内部再绑定自己的工具列表这样模型每轮的候选工具只有 3-5 个准确率和响应速度都更好。这种做法的开销是路由节点需要额外一次 LLM 调用但对于复杂 Agent 场景这点开销换来的稳定性和可维护性完全值得。5. 实操过程与核心环节实现本章给出一个可以直接抄作业的 LangGraph 多 Server 调用实现包含两个虚拟 Server一个提供文件查询工具一个提供时间工具。为了控制篇幅我只贴核心片段完整逻辑会写清楚。5.1 多 Server 初始化顺序与搬运工具先准备好两个 Server 的启动参数然后创建一个context管理函数来同时初始化多个会话import asyncio from contextlib import asynccontextmanager from mcp import ClientSession, StdioServerParameters from mcp.client.stdio import stdio_client SERVERS { file: StdioServerParameters(commandsys.executable, args[file_server.py]), time: StdioServerParameters(commandsys.executable, args[time_server.py]), } asynccontextmanager async def create_mcp_clients(): sessions {} for name, params in SERVERS.items(): read, write await stdio_client(params).__aenter__() session await ClientSession(read, write).__aenter__() await session.initialize() sessions[name] session try: yield sessions finally: for name, session in sessions.items(): await session.__aexit__(None, None, None)这个初始化顺序有讲究一定要先 initialize再 list_tools否则部分 Server 可能因为状态未就绪返回空工具列表。我见过不少代码在创建 session 后立刻 list_tools结果偶尔拿到空列表后来才发现是initialize 和 list_tools 并发执行导致时序竞争。严格串行初始化并把 session 存入字典是最笨但最可靠的方式。5.2 执行工具调用并处理结果LangGraph 节点内部LLM 返回一个 tool_calls 数组每个元素包含 name、args、id。我们需要根据工具名前缀找到对应 Server然后调用 call_tool 方法。下面是核心执行函数def make_agent_node(sessions, llm): tools [] for server_name, session in sessions.items(): mcp_tools await session.list_tools() for mcp_tool in mcp_tools.tools: tools.append(convert_mcp_tool_to_langchain(mcp_tool, server_name, session)) llm_with_tools llm.bind_tools(tools) async def agent_node(state): messages state[messages] response await llm_with_tools.ainvoke(messages) return {messages: [response]} return agent_node然后执行节点逐个处理工具调用async def tool_node(state, sessions): last_message state[messages][-1] outputs [] if not hasattr(last_message, tool_calls): return {messages: state[messages]} for tool_call in last_message.tool_calls: tool_name tool_call[name] server tool_name.split(_, 1)[0] session sessions[server] result await session.call_tool( tool_name.split(_, 1)[1], tool_call[args] ) text_parts [item.text for item in result.content if item.type text] outputs.append(ToolMessage(content\n.join(text_parts), tool_call_idtool_call[id])) return {messages: outputs}这段代码里两个细节容易踩坑。一个是 ToolMessage 的 tool_call_id 必须和原始 tool_call 的 id 一致否则模型无法关联工具结果。另一个是 call_tool 返回的 content 可能包含 image、audio 等非文本块聚合文本时必须过滤 item.type text否则会拼出大量空格和难懂的对象字符串。5.3 流式输出与并发LangGraph 天然支持状态流式但它默认不流式输出 MCP 工具的中间结果因为工具调用通常是一次性返回。如果你想让 Server 的进度通过 SSE 或 stdio 流式传出需要在 Server 端实现 notifications 或者用长耗时任务的进度通知机制。目前官方 SDK 对工具执行过程内的流式支持还在演进常规做法依然是工具快速返回最终结果LLM 再流式生成最终答案。并发方面最需要注意的是避免多个节点共享同一个 ClientSession 并发调用call_tool。MCP 规范并不强制 Server 处理并发请求Stdio 子进程往往不处理并行调用轻则排队重则直接崩。我的建议是在每个节点内部加上信号量控制并发上限或者干脆保持一个 Session 同一时刻只处理一个 call_toolsemaphore asyncio.Semaphore(1) async def safe_call_tool(session, name, args): async with semaphore: return await session.call_tool(name, args)这个简单限制能避免 90% 的随机性错误。6. 踩坑记录协议版本、超时、工具 schema 校验这一章是真正值钱的部分。我在把 MCP 嵌入 LangGraph 的过程中踩过一堆坑整理成速查表按频率排序。6.1 transport 不匹配 / 服务器退出现象客户端 initialize 阶段直接超时甚至报 “Server closed the connection without sending a response”。排查第一步确认 Server 是否真的在运行运行方式是否与启动参数匹配。如果你 command 写的是 pythonargs 写的是 server.py但当前环境里能访问的 python 是系统解释器而不是虚拟环境解释器就可能因为依赖缺失导致子进程秒退。第二步检查 stdio 是否被污染。这是个大坑如果你的 Server 代码里有任何 print 输出到 stdoutstdout 就不再是纯 JSON-RPC 消息通道客户端解析第一条消息就会失败。常见病源是调试用的 print 忘删或者第三方库往 stdout 打了日志。规范做法是所有调试信息写 stderrprint(debug log, filesys.stderr)另外Windows 上创建子进程时某些环境变量会干扰 stdio建议在 StdioServerParameters 里显式带一个干净的 env最少保留 PATH、PYTHONPATH 和系统盘符相关变量。6.2 工具调用被拒或报错现象工具能列出但 call_tool 返回 isErrorTrue。常见原因有三个参数类型不匹配、必填参数缺省、Server 内部异常。MCP 的 inputSchema 是 JSON Schema客户端传的参数要严格符合它。LangChain 的 StructuredTool 会自动做 Pydantic 校验但如果你手工构造工具函数很可能没做参数强制转换。遇到这类问题最有效的手段是给 call_tool 外层包一个 try/except把异常信息连同工具名一起返回给模型。很多情况下模型看到详细报错能自我纠错再次调用。我在生产项目里就是靠这个“错误回传重试”机制把一次调用成功率从 70% 提到 98%。6.3 服务器连接生命周期与状态污染现象图跑多轮后工具行为不一致或者状态残留。如果 Server 内部维护了会话状态比如“当前选择的数据库”那么多个不同图分支共用同一个 Server 会话就会互相污染。要避免这类问题要么把 Server 设计成无状态——每次 call_tool 都携带完整上下文要么为每个图实例创建独立的 ClientSession。我的原则是MCP Server 尽量做成纯函数式所有上下文通过参数传入不要靠服务端缓存。这样多 Server 调用才能保持可预期性。6.4 一个易被忽略的问题工具 schema 复杂度过高有些 Server 暴露的工具 inputSchema 极其复杂嵌套五六层。LLM 构造参数时很容易遗漏深层字段导致校验失败。遇到这种情况光靠模型自己猜参数不可靠。建议在工具描述里写明“参数默认值是什么、哪些字段可以省略”甚至给出一两个完整的 JSON 示例。这个描述内容会进入 prompt是调整模型调用成功率最直接的手段远比你改代码里的校验逻辑快。调试 MCP 还有一个神器建议使用MCP Inspector。它是一个图形化工具可以直接查看工具列表、修改参数调用工具、查看原始 JSON-RPC 消息。遇到诡异问题先用 Inspector 复现再回到代码里排查至少能节省一半时间。官方还有 mcp dev 可以快速启动 Server 并附加 Inspector开发体验很不错。最后多 Server 调用的稳定性归根结底取决于你的容错设计而不是协议本身。给每个 Server 调用设置超时给每次失败备好 fallback在 LangGraph 里预留一个“工具调用失败但允许模型继续”的条件分支这套思路比纠结协议版本更新更有价值。我在实际项目中就是靠这种防御式写法才敢把四个 MCP Server 同时接进一条 Agent 工作流里稳定跑数天。