我前阵子接手的一个内部自动化助手项目,逼着我把 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参数启动,日志不能混进 stdout |
| SSE (HTTP+SSE) | 客户端与服务器之间保持单向 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,模型根据全局工具清单决定要调用哪些工具;工具清单里每一项都记录了它来自哪个 Server,LangGraph 的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, env=None): params = StdioServerParameters( command=command, args=args, env=env ) 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。我先定义 State:
from 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 5) | stdio 子进程启动被权限拦截 | 检查当前终端是否以管理员权限运行;如果是从 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}, config=config)第二个是 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 的路上少走两步弯路。