我前阵子接手的一个内部自动化助手项目逼着我把 MCP 协议从握手到多 Server 调用完整摸了一遍。项目需求本身不花哨让模型能查内部知识库、能跑关系型数据库里的一张工单表、能往日程工具里塞会议还要能把结果整理成一段人话返回给用户。最初我图省事把所有工具都塞进一个 MCP Server心想一个进程解决所有问题。结果工具从 5 个涨到 15 个之后整个人都不好了。后来我把不同能力拆成了多个 MCP Server再用 LangGraph 做编排层链路才真正顺起来。这篇分享我就按这条实践路线来写先从协议握手这种底层细节讲起再讲怎么用 LangGraph 把多个 Server 组织成一个可维护的 Agent 系统。想入坑 MCP 的开发者、已经跑通单 Server 但准备上多 Server 的团队这篇应该都对得上。1. 先把“一个 Server 装下一切”这个念头打消1.1 为什么单 Server 会让系统越来越难维护很多人刚接触 MCP 时第一反应都是“一个服务把所有工具暴露给大模型”。这想法在 demo 阶段完全没问题我最初也是这么干的一个 FastMCP Server 里放了查知识库、查工单、写日历十几个工具跑得很爽。但等工具数量上来问题就开始冒头了。首先是鉴权模型搅在一起。知识库工具用的是内部 SSO token数据库工具需要独立账号密码日历工具走 OAuth2全塞在一个 Server 里初始化逻辑就变成一锅粥。其次是对单个工具的故障隔离几乎不存在一个数据库连接超时的工具崩了整个 MCP Server 的进程都跟着抖其他完全无关的工具也一起遭殃。然后是工具名冲突知识库里有search数据库查询里我也想叫search同一个 Server 里要么改名要么加前缀改到最后连自己都分不清。最后是部署升级没法独立进行哪怕只改一个 SQL 查询的逻辑也要把整个 Server 重启一遍。所以后来我的判断很清楚MCP 的协议设计本身就没要求所有工具装进同一个进程它只是定义了“客户端”和“服务端”之间的通信标准。按业务边界拆成多个 Server让每个 Server 只负责一类能力才是更接近生产可用的形态。1.2 MCP 的三层协议栈要理解多 Server 调用先得把 MCP 的协议栈看明白。我习惯把 MCP 拆成三层来看最底下是传输层中间是 JSON-RPC 2.0 消息层最上面才是大家经常提到的方法名和语义。传输层决定客户端和服务端“通过什么管道说话”。目前主流的有三种stdio、SSE、Streamable HTTP。传输方式工作方式适用场景握手要点stdio客户端启动服务端子进程通过 stdin/stdout 通信本地开发、同机部署子进程必须带--stdio参数启动日志不能混进 stdoutSSE (HTTPSSE)客户端与服务器之间保持单向 Server-Sent Events请求通过独立 HTTP endpoint 发送跨主机、需要远程访问需要先建立 SSE 连接拿到 session再用 POST 发请求Streamable HTTP单 endpoint双向消息都走 HTTP 请求/响应新版 SDK 推荐、云端部署endpoint 数量从两个收敛为一个消息通过 body 里的 JSON 编码消息层就是 JSON-RPC 2.0所有请求都形如{jsonrpc:2.0,id:1,method:xxx,params:{...}}响应里要么带result要么带error。这里的id是客户端自己维护的每个请求要有唯一 id服务端会用同一个 id 回包。协议层则定义了一系列语义化方法最常用的是initialize、tools/list、tools/call、resources/list还有prompts/list和sampling这类进阶能力。说白了传输层负责把某个 JSON 串送过去JSON-RPC 层负责请求响应配对协议层负责让双方理解“这个请求是什么意思”。理解了三层的分工后面看握手代码就完全不慌了。1.3 单 Server 到多 Server核心改变是什么从单 Server 切换到多 Server表面上是把工具拆开部署本质上是把“能力边界”从进程边界变成协议边界。客户端不再是“连接某个 Server 然后获得所有工具”而是持有多个连接每个连接对应一个独立的 Server按需拉取工具列表甚至可以让不同 Server 运行在不同的机器上。对于整个系统来说收益非常直接。每个 Server 独立鉴权知识库的 token 不会渗透到数据库 Server 的凭据体系里每个 Server 独立重启数据库工具出问题不再拖累日历工具工具名空间天然隔离search在知识库 Server 里存在在数据库 Server 里也存在互不干扰。代价就是客户端的连接管理和工具注册逻辑变复杂了而这个复杂度恰好是 LangGraph 这类编排框架能帮忙收拾的部分。2. 协议握手MCP 通信的“前台登记”2.1 initialize 到底在做什么如果你把 MCP 通信理解成“进一家公司拜访”那initialize就是前台登记环节。客户端进门先自报家门说自己是谁、从哪里来、支持到哪一版访客协议、能干什么服务端看看这些信息决定让不让你进并告诉你它这边支持哪种协议版本、能提供哪些服务范围。具体的 JSON-RPC 请求长这样{ jsonrpc: 2.0, id: 1, method: initialize, params: { protocolVersion: 2024-11-05, capabilities: {}, clientInfo: { name: my-agent, version: 0.1.0 } } }服务端收到后会继续读params里的clientInfo和capabilities再返回自己的能力声明{ jsonrpc: 2.0, id: 1, result: { protocolVersion: 2024-11-05, capabilities: { tools: {} }, serverInfo: { name: kb-server, version: 1.2.0 } } }注意这里的protocolVersion不是写死的。客户端声明自己支持到什么版本服务端在自己的支持列表里挑一个双方都能接受的返回。这块如果两边 SDK 版本差太多容易出现“版本对齐失败”表现就是握手阶段直接报错。实际开发里我建议把 SDK 依赖锁到具体小版本别用 latest 之类的大范围版本号。2.2 initialized 通知和后继请求很多初学者会在initialize拿到 result 之后立刻发tools/list结果被服务端回了一个“方法未找到”的错。原因是 MCP 的握手还没走完还差最后一步客户端需要发一个名为notifications/initialized的通知。{ jsonrpc: 2.0, method: notifications/initialized, params: {} }注意这是一个notification按 JSON-RPC 规范它没有id服务端也不需要回包。它的含义是“我已经确认了协议版本握手完成我们可以开始正常工作了”。只有在你发了这个通知之后tools/list、tools/call这类请求才会被服务端接受。tools/list本身没有握手这么复杂它就是一个普通请求服务端返回一份 schema 列表{ jsonrpc: 2.0, id: 2, method: tools/list }{ jsonrpc: 2.0, id: 2, result: { tools: [ { name: search_kb, description: 搜索内部知识库返回相关文档片段, inputSchema: { type: object, properties: { query: {type: string} }, required: [query] } } ] } }inputSchema是 JSON Schema 格式描述工具参数的结构大模型就是靠这个字段来决定怎么调用的。这块写得不清楚模型就会乱传参数。我见过有团队把description写得很随意结果模型把query传成用户 id排查半天才发现是描述歧义。2.3 不同传输模式的握手差异如果同一套 MCP Server 跑在 stdio 模式握手基本是隐式的客户端直接 fork 子进程通过 stdin 写initialize从 stdout 读响应。这个模式最干净因为进程生命周期由客户端控制所有交互都在本机。但有一点经常踩坑子进程的日志千万别打到 stdout否则会混进协议数据流导致客户端解析 JSON 失败。我一般把日志全部重定向到 stderr 或者文件里。SSE 模式就不一样了。客户端需要先建立一条 SSE 长连接这条连接是服务端到客户端的单向通道服务端会在这条连接上推送事件包括握手响应和后续的工具执行结果。而客户端真正要发请求时是向另一个 HTTP endpoint通常是/messages发 POST 请求。所以 SSE 模式的握手实际上分两条路径先走 GET 建立事件流拿到服务端下发的 endpoint 信息再往这个 endpoint 发initialize请求响应则从事件流里收。新版 SDK 推荐的 Streamable HTTP 则把两条路径收成一个 endpoint客户端发 POST响应直接走 HTTP response 返回不再需要单独维护 SSE 长连接。对部署来说省心很多特别是放在反向代理后面时不用再单独配 SSE 的 streaming 超时。3. LangGraph 多 Server 调用的架构拆解3.1 一个具体的多 Server 场景当我们讨论“LangGraph 多 Server 调用”时最好先有一个具体场景。我自己的项目是一个“工单 知识库 日程”三合一的自动化助手用户输入一句自然语言比如“客户报了个数据库连接超时的工单帮我查一下相关文档然后安排明天上午复审”。这个任务实际会触发三个服务kb-server内部知识库检索提供search_kb、get_doc_detail两个工具db-server连接业务工单库提供query_ticket、update_ticket_status工具cal-server读写团队日历提供create_event、get_availability工具。这样拆分之后每个 Server 内部只关心自己那一类数据源工具的 inputSchema 也特别干净。而编排层要做的就是先让模型理解用户意图再决定要不要依次调用search_kb、query_ticket、create_event最后把结果汇总成回复。3.2 LangGraph 为什么适合做编排层在引入 LangGraph 之前团队里可能已经有了一条朴素的“Agent 循环”让大模型自己决定调什么工具然后 while 循环一直执行直到模型说“完成”。这种方式在工具少的时候没问题工具一多、步骤一长控制力就不够了容易出现模型反复调用同一个工具、陷入死循环的情况。LangGraph 的思路是把 Agent 流程画成一张显式的图。节点是具体的处理逻辑比如“意图识别”“查工单”“查知识库”“写日历”边是流程走向条件边则让图能根据中间结果动态选择下一跳。相比“黑盒循环”图结构的好处是每个节点干了什么、什么时候结束、状态怎么流转全都可以观测。而且 LangGraph 自带检查点机制可以把每一步的状态保存下来中途失败能回放调试。还有一个很实际的原因多 Server 的工具调用天然存在依赖关系。比如你得先查工单拿到客户问题描述才能带着问题描述去搜索知识库你也要先看日历空档才能决定把会议安排在几点。这种“先 A 后 B”的顺序用条件边控制就特别自然。3.3 整体架构分层我把这套系统拆成了四层客户端连接层、工具注册层、工作流编排层、执行层。层级职责说明客户端连接层负责与多个 MCP Server 建立连接维护生命周期每个 Server 一个连接实例统一管理重连和关闭工具注册层从各 Server 拉取工具列表做前缀与 schema 归一化给每个工具打上serverName标签形成全局工具清单工作流编排层定义 LangGraph 的 State、节点、条件边模型只感知到扁平化的工具列表但实际上背后连接多个 Server执行层通过 LangGraph 节点调用 MCP 工具处理返回结果负责把 MCP 的原始返回转成结构化数据写入 State一次完整请求的路径长这样用户输入进入 LangGraph 的plan_node模型根据全局工具清单决定要调用哪些工具工具清单里每一项都记录了它来自哪个 ServerLangGraph 的tool_exec_node根据这个标签把请求路由到对应 Server各 Server 分别执行完后把结果写回共享 State最后answer_node把 State 里的所有结果汇总生成回复。整个过程对用户是透明的但工程上每一步都可以监控。4. 核心代码实现连接多 Server 与工具注册4.1 依赖准备代码我用 Python 生态来做核心依赖是这几个pip install langgraph langchain langchain-mcp-adapters mcp需要注意MCP 的 Python SDK 版本迭代很快我用的版本接口可能过一段时间会小有变化但核心思路不变。如果你照抄报错优先去查对应版本的 changelog。4.2 多 Server 客户端聚合第一步是先封装一个函数用来连接多个 Server。最简单的方式是给每个 Server 建一个stdio_client然后各自走一遍 initialize 和 tools/list 流程。示例代码如下import asyncio from mcp import ClientSession, StdioServerParameters from mcp.client.stdio import stdio_client async def connect_server(name, command, args, envNone): params StdioServerParameters( commandcommand, argsargs, envenv ) read, write await stdio_client(params).__aenter__() session await ClientSession(read, write).__aenter__() await session.initialize() tools await session.list_tools() # 给每个工具打上 Server 名前缀 return name, session, [ { **tool.model_dump(), name: f{name}_{tool.name}, server: name, } for tool in tools.tools ] async def connect_all(): servers { kb: (python, [kb_server.py]), db: (python, [db_server.py]), cal: (python, [cal_server.py]), } results await asyncio.gather( *[connect_server(n, cmd, args) for n, (cmd, args) in servers.items()] ) sessions {n: s for n, s, _ in results} all_tools [t for _, _, tools in results for t in tools] return sessions, all_tools这里的核心是两个点一是session.initialize()必须调用二是工具名加前缀。前缀很重要因为不同 Server 的工具名可能重复而且模型看到一个带前缀的工具名能更容易理解这个工具来自哪个域。我在项目里直接用kb_search_kb这种格式虽然有点冗余但模型调用准确率明显更高。顺带说一句stdio 模式连接远程部署的 Server 不方便如果你的 Server 跑在另一台机器上就得换成streamable_http_client这类基于 HTTP 的传输。接口类似换成对应的 client 即可。4.3 工具命名空间与 schema 冲突处理工具名加前缀只是第一步更麻烦的是inputSchema冲突。举个例子kb-server 和 cal-server 都有一个参数叫query但语义完全不一样知识库的 query 是搜索词日历的 query 可能是时间段。如果不做处理模型很可能把同一个参数传给两个 Server导致日历查询直接报错。我的处理方式是在工具注册层做一层归一化把description里补充明确来源同时在 LangGraph 的 tool 包装上明确注释from langchain_mcp_adapters.tools import load_mcp_tools async def build_tools(session, prefix, tool_names): raw_tools await load_mcp_tools(session) return [ { name: f{prefix}_{t.name}, description: f[{prefix}] {t.description}, args_schema: t.args_schema, func: t.func, } for t in raw_tools if t.name in tool_names ]如果某个 Server 暴露的工具数量很多你还可以在这一层做过滤只把当前工作流需要的工具注册进来。比如日历 Server 可能提供了 10 个工具但当前智能体只需要create_event和get_availability就只注册这两个。这样做的好处是模型面临的工具列表更短决策更快也不会误调高风险工具。4.4 用 LangGraph 组织调用流程注册好工具之后下一步就是把它们编排进 LangGraph。我先定义 Statefrom typing import TypedDict, Annotated from langgraph.graph import StateGraph class AgentState(TypedDict): user_input: str tool_results: Annotated[list, operator.add] final_answer: str然后定义几个节点。第一个节点是plan_node它让模型看用户输入结合全局工具列表决定要不要调用工具第二个节点是tool_node它执行实际的工具调用把结果放进tool_results第三个节点是answer_node它读取所有结果生成最终回复。 LangGraph 最核心的是条件边的构建比如根据plan_node输出的should_call_tools判断是否进入工具调用还是直接跳回用户from langgraph.graph import END def should_continue(state): if state[should_call_tools]: return tool_node return answer_node graph StateGraph(AgentState) graph.add_node(plan_node, plan) graph.add_node(tool_node, execute_tools) graph.add_node(answer_node, generate_answer) graph.add_edge(plan_node, tool_node) graph.add_conditional_edges( plan_node, should_continue, {tool_node: tool_node, answer_node: answer_node} ) graph.add_edge(tool_node, answer_node) graph.add_edge(answer_node, END) app graph.compile()这里有个容易被忽略的细节tool_node里的工具函数是异步的还是同步的。MCP 客户端大多数是异步接口如果你在plan_node里直接调用会卡住事件循环。我建议在tool_node里统一用asyncio.run()或者直接把节点定义成 async 函数LangGraph 是支持 async 节点的。5. 连 Shutdown 都能排查的常见问题清单5.1 初始化阶段的典型问题多 Server 环境里半数问题都出在握手阶段。我把最常见的几类整理成一个速查表现象可能原因排查思路initialize 请求发送后一直没有响应子进程启动失败、路径错误、服务端未开启 stdin 模式手动用命令行启动 Server确认能正常输出 JSON 日志检查 stdout 是否被业务日志污染握手后发 tools/list 报 method not found忘记发notifications/initialized在 initialize 响应后补发通知再发后续请求protocolVersion 不匹配客户端和服务端 SDK 版本差距过大统一锁版本优先对齐到同一 SDK 版本不要盲目升级tools/list 返回空数组服务端没有注册任何工具或注册工具函数时报错单独跑 Server调用 tools/list 看返回检查 Server 日志里的异常堆栈权限拒绝类报错比如 Windows 下 OS error 5stdio 子进程启动被权限拦截检查当前终端是否以管理员权限运行如果是从 IDE 启动确认 IDE 子进程权限一致我在 Windows 上踩过一次比较玄学的坑直接命令行跑 Server 没问题但放到 IDE 里启动就报拒绝访问。后来发现是 IDE 继承了管理员权限而子进程要调用的某个本地服务不允许继承空权限。遇到这种问题要记得环境和启动方式不同很可能不是代码问题而是进程上下文的问题。5.2 工具调用阶段的典型问题握手过了工具调用阶段还会冒出一堆坑这里挑几个常见的讲。工具名冲突是最常见的。你以为每个 Server 的工具名都唯一实际上一套系统里search、get_status、list_items满地都是。没有前缀策略的话模型甚至会把 A Server 的get_status当成 B Server 来调用返回结果完全错乱。加前缀和 description 标注能缓解大部分问题。返回体解析失败也是高频问题。MCP 工具的返回content字段默认带type: text业务数据往往以 JSON 字符串形式藏在这段文本里。如果你在 LangGraph 节点里直接把整段文本写入 State后续节点做结构化提取时会很痛苦。我习惯在tool_node里加一层解析把 text 内容 json.loads 成对象然后再放进 State 的tool_results。工具返回内容过大也值得单独处理。MCP 的 text 内容理论上可以很大一次返回几千行日志整个 State 瞬间膨胀。我的做法是在工具执行后做截断或摘要例如只保留前 2000 个字符超出部分提示“内容过长已省略可按需继续查询”。省下的 token 对整个 Agent 的成本影响很明显。5.3 LangGraph 编排层的坑LangGraph 本身稳定但多 Server 场景下有几个坑比较隐蔽。第一个是无限循环。模型为了“再确认一下”可能会反复调用同一个工具。你需要在编译图的时候设置recursion_limit比如app graph.compile() config {configurable: {thread_id: demo-1}, recursion_limit: 10} result app.invoke({user_input: user_input}, configconfig)第二个是 State 污染。LangGraph 的 State 默认是共享的如果多个节点同时往一个 key 里写数据后面的会覆盖前面的。这也是我为什么用Annotated[list, operator.add]来声明tool_results让多个工具结果能 append 而不是互相覆盖。第三个是并行调用时的工具超时。多个 Server 的响应速度差异很大知识库可能 200ms 返回数据库查询可能要 5 秒。如果tool_node里是顺序调用整个流程会很慢。我建议在有余力时用asyncio.gather把并行度拉上去但要注意每个 Server 的连接本身就是独立的不存在跨 Server 锁的问题。5.4 我的排查顺序如果整套系统跑不通我一般按这个顺序排查先验证单个 Server 能不能单独工作。单独连接 kb-server手动调用工具看返回结构是否符合预期。这一步能排掉大多数 Server 自身的问题。然后加第二个 Server验证两个 Server 注册进同一个工具清单后模型能否正确选择各自工具。这一步主要验证前缀策略和 schema 归一化是否有效。最后才接 LangGraph重点验证状态管理和条件边是否按预期流转。顺序不对的话你会在一个三四个 Server 的系统里面临“所有地方都像有问题”的困境排查效率极低。6. 最后分享一个小建议踩了这么多坑之后如果只保留一条经验我会建议团队一句话先把单 Server 跑通再上多 Server先手动验证握手再引入 LangGraph。很多人一上来就铺开四五个 Server 的老架构结果模型不停调错工具、状态反复污染最后整个项目被归咎于“MCP 不成熟”。其实 MCP 本身只是一个传输协议真正决定系统复杂度的是你怎么拆分边界、怎么管理工具清单、怎么设计状态流转。我个人现在最顺手的套路是每个 Server 只负责一个数据域工具列表控制在 3 到 8 个以内客户端层把工具统一加前缀LangGraph 层只感知一份扁平清单任何新 Server 接入前先用一段几十行的脚本单独验证连接和工具调用再并进编排层。这套流程跑下来后续加什么数据源都不慌了。希望这篇分享能帮你在 MCP 和 LangGraph 的路上少走两步弯路。