前几天一个做客户端的朋友问我,他手上的桌面端已经接了四个 MCP 服务器:一个管本地文件检索,一个管数据库查询,一个管内部工单系统,还有一个是自研的代码分析器。刚开始两个的时候还能忍,每个服务器建一个ClientSession、一个AsyncExitStack,工具列表分开存;到第四个的时候,他的路由代码变成了三层嵌套的 if-elif,关闭逻辑还漏了一个会话,导致子进程一直挂着不退。这篇文章就聊 python-sdk 里的ClientSessionGroup,也就是会话组——它干的事情很朴素:把多个 MCP 服务器的连接、初始化、工具/提示词/资源聚合、调用路由、生命周期回收,全部收进一个对象里。你把它当成"管理多个 MCP 服务器会话的聚合器"就行。适合正在写 MCP 客户端、Agent 编排层、或者内部工具平台的同学,不管你是刚跑通第一个 hello-world,还是已经在处理十个服务器的命名冲突,下面的内容应该都能直接拿去用。

1. 会话组到底解决什么问题

1.1 手工管多个 ClientSession 的真实代价

先把"不用会话组"的写法摊开看,才知道它省了什么。通常一个 MCP 客户端连一个服务器,标准姿势是两层上下文:transport 一层(stdio 或 SSE),ClientSession一层,然后await session.initialize(),再list_tools()。这套流程写一遍很清晰,写四遍就开始出问题了。

第一个问题是资源栈的嵌套深度。你要是老老实实写四层async with,缩进能推进到屏幕外面去;要是手动AsyncExitStack,__aenter__和__aexit__的配对就得自己盯着。我见过最常见的事故是:代码在初始化第三个服务器时抛异常,前两个的 exit stack 没有被aclose(),于是 stdio 启动的两个 Python 子进程变成孤儿进程,跑一天能把内存吃满。

第二个问题是工具列表的归属关系丢失。四个服务器各自返回一个ListToolsResult,你把所有Tool对象extend进一个大列表交给模型。模型选了一个叫search的工具,你要反查它属于哪个服务器——大多数人的做法是在工具名前面硬拼一个前缀,然后调用时 split 一下。拼前缀本身没错,错在拼的时机和规则是散落在各处的字符串操作,服务器一多就必然有人拼错分隔符。

第三个问题是调用路径分叉。不同服务器的call_tool超时策略、错误语义、参数校验强度都不一样。数据库查询要 30 秒超时,文件检索 5 秒就够,工单系统可能还要带认证 header。当你只有session.call_tool一个入口时,这些差异只能靠调用方到处传参,最终写成一堆if server == "db": timeout = 30。

这三个问题叠加起来,本质是缺少一个统一的聚合层:谁负责建连、谁负责记账、谁负责路由、谁负责收尾,没有明确分工。

1.2 会话组的设计取舍:为什么是"继承"而不是"包装"

ClientSessionGroup的实现在设计上做了一个挺关键的选择:它同时继承ClientSession和AsyncExitStack。这个选择值得说道说道,因为理解了它,你就理解了整个类的行为边界。

继承AsyncExitStack解决的是资源回收。会话组自己就是一个 exit stack,connect_to_server每成功建一个子会话,就把它的清理函数push进栈里。好处是所有子会话的生命周期天然按 LIFO 顺序回收——后连的先关,避免"依赖方先死、被依赖方还活着"的尴尬。你在会话组外面套一层async with,退出时整条链自动收干净,不需要手工维护 session 列表再倒序aclose()。

继承ClientSession解决的是接口一致性。会话组对外暴露的方法名和单个ClientSession基本对齐:list_tools、call_tool、list_prompts、get_prompt、list_resources、read_resource、list_resource_templates。这意味着你原来针对单会话写的业务代码,把变量从session换成group之后大部分能直接跑。这个设计的意图很明确:让"一个服务器"和"一组服务器"在使用者眼里没有区别,降低迁移成本。

提示:接口对齐不等于语义对齐。单会话的list_tools一定是那次网络请求的实时结果;会话组的list_tools多数版本返回的是聚合缓存,带有明显的"快照"性质。这一点在第 5 章会展开讲,是排查问题时最容易踩的坑。

代价也有:因为它是ClientSession的子类,一些父类方法和聚合方法同名但行为不同,读源码时容易混淆;另外多重继承带来的 MRO 顺序在调试时不算友好。但对使用者来说,这些代价基本被隐藏了。

1.3 什么场景该用,什么场景别硬上

会话组不是万能胶水,判断标准其实很清晰。

适合用的场景:客户端要同时接 3 到 10 个服务器,且这些服务器的工具在语义上可以放进同一个"工具池"给模型挑。典型如桌面 AI 客户端、内部研发助手、需要同时读本地文件和查线上数据的分析工具。另一个很适合的场景是开发调试:你想对比两个服务器同名工具的行为差异,会话组一次连上,打印两份 schema,一目了然。

不建议用的场景:服务器数量上到几十上百,这时候你需要的不是聚合,而是服务发现加懒加载,把所有工具一次性拉进上下文会直接把 token 打爆;或者各个服务器需要完全独立的故障隔离和重连策略(比如一个连不上不能影响另一个的初始化),会话组共享一个 exit stack,一个子会话挂掉之后的补偿逻辑要你自己写;还有需要让模型明确知道"当前这个工具来自哪个可信域"的强隔离场景,光靠名字前缀是不够的,得在调用层做权限校验。

一句话总结:会话组管的是"聚合与路由",不管"治理与策略"。策略层得你自己包一层。

2. 核心 API 拆解:一个对象背后的四件事

2.1 构造参数与生命周期管理

照例先给自查手段。MCP python-sdk 迭代很快,不同小版本的签名有出入,与其照抄我的代码然后报TypeError,不如先花十秒确认一下你装的那个版本长什么样:

import inspect from mcp import ClientSessionGroup print(inspect.signature(ClientSessionGroup.__init__)) print(inspect.signature(ClientSessionGroup.connect_to_server)) print(inspect.signature(ClientSessionGroup.call_tool))

这三行输出的信息量比任何文档都准。我个人的习惯是把这段塞进一个_diag.py,每次升级 SDK 先跑一遍,把输出贴在项目的 UPGRADE 笔记里,后面出问题一眼就能看出是签名变了还是我代码写错了。

构造函数通常接受两个参数:exit_stack(可选的AsyncExitStack,默认自己新建一个)和component_name_hook(可选的名字处理回调,后面单独讲)。exit_stack这个口子有实际用途:如果你的应用已经有一个全局的 exit stack 管着数据库连接池、HTTP 客户端之类的东西,可以把会话组挂进去,统一在同一处回收,避免出现"应用关了但 MCP 子进程还在"的情况。

from contextlib import AsyncExitStack from mcp import ClientSessionGroup app_stack = AsyncExitStack() group = ClientSessionGroup(exit_stack=app_stack)

生命周期上有个容易忽略的点:会话组自身也要initialize。因为它是ClientSession的子类,外面套async with ClientSessionGroup() as group:时,进入和退出会调用父类的__aenter__/__aexit__,完成协议层的初始化握手。如果你是手动connect_to_server而不走async with,记得显式处理这一步,否则某些版本上子会话的initialize会因为父会话状态不对而失败。

2.2 connect_to_server 与传输参数

connect_to_server是整个类的入口,它接收一个描述服务器的参数对象,返回一个建好的子ClientSession。返回值的用途是给你留一个"后门":某些只对单个服务器有意义的操作(比如读它特有的 prompt、单独查询它的连接状态),可以直接在这个子会话上做,不必绕道聚合层。

参数对象按传输类型分几种:StdioServerParameters(本地子进程)、SSE 相关参数、以及流式 HTTP 相关参数。stdio 那套最常用,核心字段就四个:

env和cwd这两个字段我要多说两句。stdio 启动的子进程默认环境是继承来的,但很多部署环境(容器、IDE 插件宿主、打包后的桌面应用)会清洗环境变量,导致本地调试好好的服务器在集成后报"找不到凭据"。可靠做法是把关键变量显式塞进env,而不是祈祷继承成功。cwd的问题更隐蔽:服务器代码里写了open("config.json"),你在项目根目录跑没问题,客户端从别的目录启动就找不到文件。这类问题排查起来极费时间,提前填好cwd能省掉。

还有服务器名字。聚合层需要给每个服务器一个标识,用于日志、名字冲突处理和反查。多数版本的参数对象支持一个name字段,不填则由实现生成一个默认名(通常是命令名加序号之类)。我的建议是永远显式填,理由很实际:默认名的生成规则会随版本变,而你写在 hook 和日志里的名字不会变;一旦默认名改了,你所有的前缀规则一起失效。

2.3 聚合缓存:tools、prompts、resources 三个字典

连上服务器之后,会话组内部维护了几个字典,把组件名映射到组件对象。通常能看到tools、prompts、resources、resource_templates,外加一个sessions记录子会话。这几个字典是理解整个类的钥匙。

工作流程大致是:每次connect_to_server成功,内部会调用list_tools/list_prompts/list_resources拉一遍全量组件,经过component_name_hook处理后写进字典;同时建立"组件名 → 所属 session"的反查关系。之后call_tool(name, args)就是查表、转发、返回三步。

这带来两个直接影响。第一,连接是有成本的。连一个装了二十个工具的服务器,握手之后还要跑三次 list 请求,如果你并发连五个服务器,启动阶段的请求量是实打实的。我在一个冷启动敏感的场景里测过,五个 stdio 服务器串行连接加聚合大约 1.2 到 2 秒,主要时间花在 Python 子进程启动和三次 list 往返上,改成anyio并发连接能压到 600 毫秒左右,但要注意并发连接时某些服务器会抢同一个端口或同一个锁文件。

第二,缓存意味着快照。字典里存的是连接那一刻的工具列表。服务器后来动态注册了新工具(热重载、插件加载),会话组不会自动知道。如果你需要刷新,通常的做法是断开重连,或者直接操作子会话重新 list 再手工更新。这一点在设计插件化服务器时必须有意识。

2.4 名称冲突:component_name_hook 才是设计的重点

多服务器聚合最棘手的问题就是重名。两个服务器都提供search,三个都提供read_file,这不是夸张,是常态——因为工具命名都遵循相似的直觉。

会话组的解法是把命名权交给用户:提供一个component_name_hook,签名大致是"服务器名 + 组件名 + 组件类型 → 最终名字"。默认实现通常是给组件名加上服务器名前缀,分隔符各版本可能是一个下划线或者点号,别猜,打印出来看:

for tool in (await group.list_tools()).tools: print(tool.name)

看到实际输出之后,你才知道调用时该传什么。这一步看起来傻,但能省掉一次"我明明写了search为什么报 tool not found"的困惑。

自定义 hook 的价值不只是解决冲突,更是建立命名空间语义。比如内部系统我习惯用sys_前缀标记可信域,用user_标记来自用户自建服务器的工具,方便在调用前做策略判断:

TRUSTED = {"filesystem", "database"} def ns_hook(server_name: str, component_name: str, component_type: str) -> str: prefix = "sys" if server_name in TRUSTED else "user" return f"{prefix}__{server_name}__{component_name}"

这个命名规则有讲究:用双下划线做分隔符,是因为单个下划线在工具名里太常见(read_file、get_user),用__分割在解析时歧义最小;前缀放最前面,是为了让工具列表按可信域自然排序,模型看列表时能形成分组直觉。

注意:hook 是全局的,对 tools、prompts、resources 都会生效。如果三个类型的命名习惯差异很大(比如 resource 用 URI 风格),在 hook 里按component_type分支处理,别一套规则套到底。

2.5 路由调用与错误传播

call_tool的内部逻辑很简单:查字典拿到目标子会话,把参数原样转发,等结果。返回值是标准的CallToolResult,里面有content列表和isError标志。

真正需要你关心的是错误从哪来、到哪去。可能的失败点至少有四层:名字查不到(本地字典 miss)、子会话已经断开(transport 层错误)、服务器端工具执行抛异常(通常包装成isError=True的结果而不是异常)、以及超时。前三层的表现完全不同,如果你不区分,日志里就只剩一片 "tool call failed"。

我的处理习惯是包一层薄封装,把失败点标出来:

from mcp.shared.exceptions import McpError async def safe_call(group, name, args, timeout=None): if name not in group.tools: raise KeyError(f"未注册的工具: {name},当前可用: {sorted(group.tools)[:10]}") try: result = await group.call_tool(name, args, read_timeout_seconds=timeout) except McpError as e: raise RuntimeError(f"[{name}] 协议层错误: {e.error.message}") from e if result.isError: # 服务器端业务异常,把内容拼出来便于定位 text = "".join(getattr(c, "text", "") for c in result.content) raise RuntimeError(f"[{name}] 工具执行失败: {text[:200]}") return result

这段代码里我认为最有价值的是那个KeyError的提示语——把可用工具的前十个塞进异常信息。模型调错工具名的时候,你在日志里能直接看到候选列表,不用再去翻服务器源码。read_timeout_seconds这个参数在多数版本的call_tool上是支持的,如果不支持,用anyio.fail_after在外面兜一层也一样。

3. 从零跑通:聚合两个 MCP 服务器

3.1 环境准备与版本锁定

先说依赖。最小集是 SDK 本体加一个异步运行时,stdio 传输依赖的是标准库子进程,不需要额外装东西:

python -m venv .venv source .venv/bin/activate pip install "mcp[cli]" pip freeze | grep -i mcp

把版本号记下来。MCP 生态现在还在快速演进,ClientSessionGroup这类聚合层 API 在不同小版本之间的字段名变化不算罕见,项目里最好把版本钉死,升级时跑一遍 2.1 节那段自查脚本。

如果你要用 SSE 或流式 HTTP 传输,确认环境里有对应的 HTTP 客户端依赖;如果你打算用FastMCP快速造测试服务器,mcp[cli]这个 extras 会一起装上。目录结构我一般这么放:

demo/ servers/ weather_server.py notes_server.py client.py _diag.py

两个服务器分开放,方便单独启动调试——完全可以在终端里python servers/weather_server.py手动跑一下,确认它能响应 stdio 握手,再交给会话组。这个"先单测再聚合"的习惯能挡掉八成问题。

3.2 搭两个最小可用的服务器

用FastMCP写,几行就能出来。第一个是天气服务:

# servers/weather_server.py from mcp.server.fastmcp import FastMCP mcp = FastMCP("weather") @mcp.tool() def get_forecast(city: str) -> str: """查询城市天气""" return f"{city}: 晴,26 摄氏度,东南风 2 级" @mcp.tool() def list_cities() -> str: """列出支持的城市""" return "北京,上海,杭州,深圳" if __name__ == "__main__": mcp.run()

第二个是笔记服务,故意也叫一个容易撞名的工具,用来演示冲突处理:

# servers/notes_server.py from mcp.server.fastmcp import FastMCP mcp = FastMCP("notes") _store: dict[str, str] = {} @mcp.tool() def save_note(title: str, body: str) -> str: """保存一条笔记""" _store[title] = body return f"已保存: {title}" @mcp.tool() def list_notes() -> str: """列出已有笔记标题""" return ", ".join(_store) or "(空)" if __name__ == "__main__": mcp.run()

两个服务器都是纯 stdio,无外部依赖,拿来验证聚合逻辑足够了。真实项目里你会遇到带认证、要连数据库、启动慢到十几秒的服务器,但排错时先保证最小集合能跑通,再逐个加复杂度,这是我一直用的方法。

3.3 用会话组一次连上两个服务器

客户端代码:

# client.py import anyio from mcp import ClientSessionGroup, StdioServerParameters SERVERS = [ ("weather", "servers/weather_server.py"), ("notes", "servers/notes_server.py"), ] async def main() -> None: async with ClientSessionGroup() as group: for name, script in SERVERS: params = StdioServerParameters( command="python", args=[script], cwd=".", env={"PYTHONUNBUFFERED": "1"}, ) session = await group.connect_to_server(params) print(f"[ok] 已连接 {name} -> {len(group.tools)} 个工具累计") print("\n=== 聚合后的工具列表 ===") for tool in (await group.list_tools()).tools: print(f"{tool.name:<40} {tool.description or ''}") print("\n=== 调用 ===") r1 = await group.call_tool("get_forecast", {"city": "杭州"}) print("天气:", r1.content[0].text) r2 = await group.call_tool("save_note", {"title": "mcp", "body": "会话组测试"}) print("笔记:", r2.content[0].text) if __name__ == "__main__": anyio.run(main)

跑之前有一件事必须确认:group.tools里的键到底叫get_forecast还是weather_get_forecast。上面的call_tool我按不带前缀写的,如果你的环境跑出来tool not found,把工具列表打印出来,照着实际名字改调用参数即可。这就是我在 2.4 节强调"别猜,打印"的原因——这一步花十秒,比对着报错猜十分钟值。

还有一个细节:env里我只传了PYTHONUNBUFFERED。如果你需要给服务器传凭据,记住 stdio 的父子进程是同一台机器、同一个用户,环境变量里塞密钥是可以的,但注意不要把它打进日志。生产环境更稳妥的做法是服务器自己从配置中心读,客户端只传一个指向配置的标识。

3.4 参数选择背后的几个数字

代码跑通了,但工程上还得回答几个"多少"的问题,这些文档里通常不写,只能靠经验和测量。

子进程内存占用。每个 stdio 服务器就是独立进程。一个只依赖标准库的FastMCP服务器,常驻大约 40 到 60 MB;如果服务器自己加载了 pandas、向量库、大模型 tokenizer 这类东西,单进程 200 MB 到 1 GB 都正常。所以"十个服务器"在内存账上不是小事:轻量工具类大概 500 MB 量级,吃数据的能到几 GB。设计阶段就得判断,哪些服务器该懒加载——不用的时候别连。

并发连接数。connect_to_server是可以并发的,但并发度建议控制在 3 到 5。原因不是客户端扛不住,而是 stdio 子进程启动时常常要做重活(加载模型、建连接池、读大文件),同时起十个会让磁盘和 CPU 互相打抢,总时间反而更长。实测下来,3 到 5 的并发度在机械盘和普通云主机上表现比较均衡。

超时设置。会话组的call_tool超时,我按工具类型分三档:纯计算和本地读写下 5 秒;网络请求类 15 秒;涉及批量数据处理或外部系统同步的 60 秒以上。不要图省事设一个 120 秒全局值,那样一次卡死会拖住整个请求链路,而且在并发场景下会把排队时间越拖越长。

工具数量的上限感知。这一点直接关系到 token。每个工具的名字、描述、JSON Schema 都要塞进模型上下文。经验值:一个中等描述的工具大约占 80 到 150 token,二十个工具就是 2 到 3 千 token,外加系统提示词里对用法的说明。如果你聚合了五个服务器、每个十个工具,还没开始干活就已经花掉了好几千 token。所以第 4 章的"工具裁剪"不是锦上添花,是必需品。

3.5 优雅关闭:子进程别留成僵尸

关闭这件事一旦出事就很烦人。会话组的async with退出时会走 exit stack,先关子会话再关 transport,stdio 传输关闭时会终止子进程。但在三种情况下仍然可能留下残留:

第一种,初始化失败。前两个服务器连上了,第三个抛异常,async with的清理会执行,但如果你的代码在此之前又手工建了一个 exit stack(有人为了"更好的控制"这么干),两层栈的管理权就乱了。

第二种,服务器自己派生了孙进程。比如服务器内部起了一个 Node 子进程或后台 worker,主进程被终止时孙进程可能活着。这类问题在客户端侧基本无解,得在服务器侧处理信号和进程组。

第三种,进程被强杀。客户端自己SIGKILL了,什么都来不及清理,只能靠下次启动时按残留的 PID 文件做兜底回收。

我的处理办法很土但有效:开发阶段在finally里打印一下存活的子进程数,跑几轮之后如果数字只增不减,就说明清理链有漏点:

import os, subprocess def count_python_children() -> int: out = subprocess.run( ["ps", "-o", "pid,ppid,comm", "--no-headers"], capture_output=True, text=True, ).stdout return sum(1 for line in out.splitlines() if line.split()[-1].startswith("python") and int(line.split()[1]) == os.getpid())

这个函数只在 Linux/macOS 上能跑,Windows 上换个思路(用tasklist或者直接查psutil)。它不优雅,但能让你在五分钟内发现问题,而不是等到用户抱怨电脑卡。

4. 进阶:命名空间、裁剪与并发

4.1 把 hook 当成权限边界的锚点

前面给的 hook 示例是在拼前缀,这里再往前推一步:hook 产出的名字,可以成为你调用链路上的策略锚点。因为所有聚合后的组件名都经过这一个函数,你在名字里编入的语义(可信域、来源、类型)会稳定地出现在后续所有位置——日志、模型可见的工具列表、调用入参、异常信息。

具体做法是在调用前解析前缀:

BLOCKED = {"user__shell__exec"} async def guarded_call(group, name, args): if name in BLOCKED: raise PermissionError(f"该工具已被策略禁用: {name}") if name.startswith("user__") and name.endswith("__write"): raise PermissionError(f"用户自建服务器的写操作需二次确认: {name}") return await group.call_tool(name, args)

要承认这套方案的天然局限:它依赖于"名字解析正确",而不是真正的能力隔离。如果某个服务器自己注册了一个前缀恰好是sys__的工具,这条防线就形同虚设。所以真正做安全边界,还得在连接阶段决定连不连,而不是在命名阶段决定拦不拦。hook 是方便的第一道网,不是最后一堵墙。

4.2 工具裁剪:让上下文回到能用的体量

聚合之后动辄三五十个工具,模型准确率会明显下滑——不是模型不行,是候选集太大,描述之间又互相干扰。裁剪手段按效果排序:

第一档,连接层裁剪。用不上的服务器压根不连。这最彻底,但要判断"用不上"就得有场景前置知识,适合工具分支明确的客户端(比如聊天模式只连文档检索,编码模式只连代码分析)。

第二档,注册前裁剪。连上之后、把工具交给模型之前,过一遍白名单:

KEEP = { "weather__get_forecast", "notes__save_note", "notes__list_notes", } async def collect_tools(group): all_tools = (await group.list_tools()).tools kept = [t for t in all_tools if t.name in KEEP] dropped = len(all_tools) - len(kept) print(f"工具裁剪: 保留 {len(kept)},过滤 {dropped}") return kept

注意这里只影响"传给模型的那份列表",会话组内部的字典不动,所以裁剪后仍然可以按名字调用被过滤掉的工具——这一点在调试时很有用,你可以让模型只看到三个工具,但手工调用第四个来验证服务器行为。

第三档,按需展开。把工具分成"常驻"和"可发现"两组,常驻的放进初始上下文,剩下的提供一个元工具(比如list_more_tools/use_tool),模型需要时再拉。这套方案能压到极致,但实现成本高,而且多轮交互会增加延迟,适合工具集特别大的平台。

4.3 并发调用与故障隔离

会话组的call_tool本身是异步的,多个调用可以并发。用anyio.create_task_group组织,注意异常处理——task group 里任何一个任务抛异常默认会取消其他任务,如果你要"部分成功",得自己捕获:

import anyio async def batch_call(group, calls: list[tuple[str, dict]]): results = {} async def one(name, args): try: r = await group.call_tool(name, args, read_timeout_seconds=30) results[name] = ("ok", r.content[0].text if r.content else "") except Exception as e: results[name] = ("err", str(e)) async with anyio.create_task_group() as tg: for name, args in calls: tg.start_soon(one, name, args) return results

one里把异常全吃掉是关键,这样单个工具失败不会导致整批取消。但"吃掉"不等于"忽略"——结果字典里带上了错误信息,上层可以决定是重试还是降级。

故障隔离还有一个维度:某个服务器的子会话挂了,会不会影响其他服务器?答案是基本不影响,因为每个子会话有自己的 transport 和读写流。但会话组的聚合字典里可能还留着这个服务器已经失效的工具名,这时调用会走到一个断掉的会话上,报传输层错误。稳妥做法是给每个服务器的调用包一层,失败达到阈值就从你的路由表里摘掉它(注意不要直接改会话组内部字典,自己在外面维护一份可用集合更安全)。

4.4 接进自己的 Agent 循环

最后落地一下:会话组怎么和模型调用循环拼起来。核心就三步——启动时连服务器、每轮把工具列表转成模型要的格式、模型返回工具调用时路由过去。

def to_openai_tools(tools): return [ { "type": "function", "function": { "name": t.name, "description": t.description or "", "parameters": t.inputSchema, }, } for t in tools ]

inputSchema就是服务器声明的 JSON Schema,基本可以直接喂给函数调用接口,少数情况下要做归一化(比如去掉某些方言字段)。工具名长度有约束的接口要注意,hook 拼太长可能超限,那种情况下用短前缀加哈希后缀更稳。

调用回来之后,把(tool_call_id, 结果文本)按接口要求压回消息列表,继续下一轮。这里有个工程细节值得强调:结果文本要截断。工具返回几万字符的情况太常见(读文件、查日志),不截断会直接把上下文撑爆。我的默认策略是单条结果超过 4000 字符就截断并附一句"已截断,完整内容可通过参数 xxx 分页获取",具体阈值看你模型的窗口大小。

5. 常见问题与排查实录

5.1 名字相关的坑

这是最高频的一类。表现是call_tool报找不到工具,但你明明看到服务器里有。原因通常有三个:没看打印出来的实际名字(前缀规则和你以为的不一样);hook 里有分支写漏了,某类组件没被处理导致名字是空或者重复;多个服务器同名工具被后连的覆盖了——注意字典赋值是覆盖语义,如果你的 hook 返回了相同名字,先连的会静默消失。

排查顺序建议固定成:打印group.tools的所有键、检查键的数量是否等于各服务器工具数之和、单独对某个服务器await session.list_tools()看原始名字。第三步能立刻区分"服务器没提供"和"聚合层改名了"。

5.2 连接阶段的问题

连接阶段的排查心法是降维:把会话组的代码先放一边,直接用终端手动启动服务器,看它能不能正常握手;能,问题在参数;不能,问题在服务器本身。这个二分法能省掉大量无谓的代码审查。

5.3 调用阶段的问题

最容易误判的是"工具返回了错误但我以为连接断了"。isError=True的返回和抛异常是两码事:前者是服务器正常响应、业务失败;后者是链路问题。日志里要把这两种分开打标记,否则你会在网络排查上浪费时间。

第二个常见问题是参数 schema 不匹配。模型给的参数名和服务器声明的差一点(cityvscity_name),服务器返回校验错误,表现却像是"工具不好用"。缓解办法是在系统提示里明确要求严格按 schema 填,并在客户端做一次轻量预校验——用jsonschema校验一下inputSchema,把错误在本地就打出来,附带正确示例。

第三个是超时。默认超时可能比你想的短,尤其是首次调用需要预热(加载模型、建连接)的服务器。表现是第一批请求全超时,后面就正常了。给首次调用单独放大超时,或者连上之后主动做一次"预热调用"。

5.4 关闭阶段的坑与自查清单

我在项目里留了一份固定的自查清单,每次改动聚合逻辑都过一遍:

• 退出async with之后,子进程数量是否回到基线;
• 初始化中途抛异常,前序连接是否被回收;
• 同一个 group 重复connect_to_server同一个服务器,是否如预期产生两个会话(通常是,要避免);
• 服务器异常退出后,再调用它的工具报的是什么错误,是否可识别;
• 关闭顺序是否 LIFO,有没有出现"transport 先关、会话还在发请求"的报错。

最后一条特别值得注意。如果调试时看到关闭阶段刷一堆ClosedResourceError,多数是因为退出流程被打断,或者你自己在finally里又碰了一下子会话。清理阶段的原则是只关不查:不要在__aexit__里再去list_tools或者打日志调用子会话。

5.5 一份可以直接抄的排错顺序

遇到问题按这个顺序走,基本能覆盖九成情况:

1. 打印 SDK 版本和三个关键方法签名;
2. 单服务器、纯终端,确认服务器本身可用;
3. 单服务器 + 会话组,看工具名;
4. 双服务器 + 会话组,看名字冲突和前缀;
5. 打印env和cwd,确认子进程环境和本地一致;
6. 给调用加显式超时和错误分类日志;
7. 退出后查残留子进程。

这个顺序的价值在于每一步只引入一个变量。跳过第 2 步直接跳到第 4 步,是绝大多数"折腾一下午"的开端。

我在实际使用中的体会是,ClientSessionGroup这类聚合层的真正价值不在省代码行数,而在于把"连接、记账、路由、回收"这四件事收敛到了一个明确的责任主体上。当你的客户端从两个服务器长到八个,代码复杂度不是线性增长的,是往乘法走的;有个地方统一管着,出问题时你至少知道该看哪个文件。至于工具裁剪、命名策略、权限校验这些,会话组给了你钩子和字典,剩下的得按自己的业务场景一个个补——这部分没有捷径,但补完之后整个客户端会稳得多。