python-sdk 中的 ClientSessionGroup：用一组对象聚合管理多个 MCP Server 连接

• 人工智能
• MCP 服务
• MCP Clients

导读

单个Client只能连接一台 MCP server，而真实应用往往同时需要访问多个 server（如搜索服务、数据库服务、内部 API），逐个维护连接和工具清单既繁琐又容易出错。python-sdk 提供的ClientSessionGroup用一个对象容纳多条连接，并把它们暴露的所有 tools、resources、prompts 聚合到一个统一的视图（dict）中，配合component_name_hook还能优雅解决跨 server 的命名冲突。读完本文你将掌握：如何用connect_to_server聚合多个 server、如何用 hook 重写组件名称避免碰撞、如何动态增删 server，以及 group 底层采用经典initialize握手的原理。

本文对应的英文原文为 docs/client/session-groups.md，其多语言译文（含印地语）位于 i18n/hi/pages/client/session-groups.md，文中的示例代码均取自 docs_src/session_groups/ 目录，源码实现位于 src/mcp/client/session_group.py。

两个彼此独立的 Server

先从两个最简单的 server 开始。它们之间没有任何关联，因此两个 server 都不约而同地把自己的 tool 命名为search——这正是后面要解决的命名冲突问题的起点。

第一个是“图书馆”服务 docs_src/session_groups/tutorial001.py：

from mcp.server import MCPServer mcp = MCPServer("Library") @mcp.tool() def search(query: str) -> str: """Search the library catalog.""" return f"3 books match {query!r}." @mcp.resource("library://hours") def hours() -> str: """When the library is open.""" return "Mon-Fri 09:00-17:00"

第二个是“网页搜索”服务 docs_src/session_groups/tutorial002.py：

from mcp.server import MCPServer mcp = MCPServer("Web") @mcp.tool() def search(query: str) -> str: """Search the web.""" return f"12 pages match {query!r}."

注意观察两点：两个 server 的 tool 都叫search；而MCPServer("Library")/MCPServer("Web")传入的构造参数"Library"、"Web"，正是后续component_name_hook拿到server_info.name的来源，这一点在 docs_src/session_groups/tutorial004.py 中会体现出来。

一个 Group：聚合多条连接

创建ClientSessionGroup，然后为每个 server 调用一次connect_to_server。完整客户端示例见 docs_src/session_groups/tutorial003.py：

import asyncio from mcp import ClientSessionGroup, StdioServerParameters async def main() -> None: library = StdioServerParameters(command="uv", args=["run", "mcp", "run", "library_server.py"]) web = StdioServerParameters(command="uv", args=["run", "mcp", "run", "web_server.py"]) async with ClientSessionGroup() as group: await group.connect_to_server(library) await group.connect_to_server(web) result = await group.call_tool("search", {"query": "model context protocol"}) print(result.structured_content) if __name__ == "__main__": asyncio.run(main())

ClientSessionGroup支持异步上下文管理器协议，async with退出时会自动关闭它创建的 exit stack 并并发关闭所有会话的专用 exit stack（见 session_group.py 中的__aenter__/__aexit__）。在使用上，有三个关键点：

• connect_to_server接收的是传输参数（transport parameters），而不是 server 对象：要启动子进程就用StdioServerParameters（从mcp顶层导入）；要连接一个已经在 URL 上监听的 server，则用StreamableHttpParameters或SseServerParameters（从mcp.client.session_group导入）。
• group.tools是所有已连接 server 的 tools 聚合结果，类型是dict[str, Tool]；group.resources和group.prompts形状完全相同，只是元素类型分别是Resource和Prompt。三个属性在源码中分别返回内部维护的self._tools、self._resources、self._prompts字典（session_group.py）。
• group.call_tool(name, arguments)会先查名字、找到拥有该 tool 的 session，再把调用转发过去——你永远不需要自己指定“该调哪台 server”。

三种传输参数的字段说明

结合 session_group.py 源码，ServerParameters实际上是三类参数模型的联合类型（StdioServerParameters | SseServerParameters | StreamableHttpParameters，见 session_group.py）：

此外，connect_to_server还接受第二个可选参数session_params: ClientSessionParameters（session_group.py），它是一个 dataclass，可以配置read_timeout_seconds、sampling_callback、elicitation_callback、list_roots_callback、logging_callback、message_handler、client_info等会话级行为——即把单个ClientSession的初始化选项透传给 group 内部建立的每个会话。

命名冲突：整个 Group 内名字必须唯一

把client.py和两个 server 放在一起运行，第二次connect_to_server会直接拒绝：

mcp.shared.exceptions.MCPError: {'search'} already exist in group tools.

这是一个MCPError，它会在第二个 server 的任何组件被注册进聚合字典之前抛出。也就是说，一个名字必须在整个 group 内唯一；当两个 server 都不受你控制时，迟早会发生碰撞。

从源码看，这个校验发生在_aggregate_components中（session_group.py）：group 先把新 session 的 prompts、resources、tools 分别写入临时字典，然后与已有的聚合字典做键集合交集运算，一旦发现重复就抛出MCPError（错误码为INVALID_PARAMS），并且整个过程不会污染已有的聚合结果——这正是文档所说“第二个 server 的任何东西都不会被注册”的机制保障。对应测试见 tests/docs_src/test_session_groups.py 中的test_colliding_names_are_rejected，它断言异常信息与文档一致，且抛出后sorted(group.tools) == ["search"]。

component_name_hook：在 Group 层重写所有注册名

解决冲突的位置在group 上，而不是两个 server 上。传入一个接收(name, server_info)的函数，group 会对它注册的每一个名字调用这个 hook。完整示例见 docs_src/session_groups/tutorial004.py：

import asyncio from mcp import ClientSessionGroup, StdioServerParameters from mcp.types import Implementation def by_server(name: str, server_info: Implementation) -> str: return f"{server_info.name}.{name}" async def main() -> None: library = StdioServerParameters(command="uv", args=["run", "mcp", "run", "library_server.py"]) web = StdioServerParameters(command="uv", args=["run", "mcp", "run", "web_server.py"]) async with ClientSessionGroup(component_name_hook=by_server) as group: await group.connect_to_server(library) await group.connect_to_server(web) print(sorted(group.tools)) result = await group.call_tool("Web.search", {"query": "model context protocol"}) print(result.structured_content) if __name__ == "__main__": asyncio.run(main())

再次运行，print(sorted(group.tools))现在能同时看到两个 tool：

['Library.search', 'Web.search']

component_name_hook通过ClientSessionGroup(component_name_hook=...)构造参数传入，源码中的类型是Callable[[str, types.Implementation], str]（session_group.py），并在_component_name方法中作用于每个待注册名字（session_group.py）。理解它需要抓住三个要点：

• dict 的 key 是你的。示例中的by_server用server_info.name拼出 key，即每个MCPServer(...)构造时传入的名字。
• 内部的Tool对象原封不动：group.tools["Web.search"].name仍然是"search"，call_tool发到 wire 上的也是这个原始名字。前缀永远只存在于你的进程内部，不会泄漏给 server。这一行为有专门测试验证（见 tests/docs_src/test_session_groups.py 的test_the_key_is_prefixed_but_the_wire_name_is_not）。
• hook 不止作用于 tools。library 的hoursresource 会以Library.hours的名字注册进group.resources；同样的规则统一作用于 prompts、resources、tools 三类组件（见_aggregate_components中对三个列表的循环处理，session_group.py）。

注意：hook 会对每个 server 的每个名字都运行，而不仅仅在发生冲突时运行。SDK 没有“仅在冲突时加前缀”的模式，因此请选择一种命名方案并让它全局生效。

call_tool 的路由原理

聚合之后，group.call_tool是如何知道该把"Web.search"转发给哪台 server 的？源码维护了一张反向索引_tool_to_session: dict[str, mcp.ClientSession]（session_group.py），在_aggregate_components聚合 tools 的同时填充（tool_to_session_temp[name] = session）。调用时（session_group.py）：

session = self._tool_to_session[name] session_tool_name = self.tools[name].name return await session.call_tool(session_tool_name, ...)

先按名字找到拥有者 session，再取出Tool的真实名称（可能是被 hook 改写前的原始名）发起调用。相关测试test_call_tool_routes_to_the_owning_server验证了Library.search与Web.search会各自命中正确的 server 并返回各自的搜索结果（tests/docs_src/test_session_groups.py）。

动态添加与移除 Server

connect_to_server会返回它打开的ClientSession。如果将来想移除某台 server，把它保存下来，然后调用：

await group.disconnect_from_server(session)

该调用会把这台 server 的 tools、resources、prompts 全部从 group 中移除。源码实现（session_group.py）依赖聚合时建立的“反向索引”_ComponentNames（记录每个 session 贡献了哪些名字，session_group.py）：断开时按索引逐个从_prompts、_resources、_tools、_tool_to_session中删除对应条目，并关闭该 session 专属的 exit stack。测试test_disconnect_removes_every_component_of_that_server验证断开后只剩Library.search与Library.hours（tests/docs_src/test_session_groups.py）。

如果你手上已经有一个连接好的ClientSession（比如Client.session就是这样一个实例），不必再开新传输，直接交给 group 即可：

await group.connect_with_session(server_info, session)

它会以同样的方式聚合这个已有会话的组件。注意两点：

• group 永远不会关闭它自己没有打开的 session。connect_with_session只做聚合（内部调用_aggregate_components，见 session_group.py），资源释放的责任仍在你手中。
• server_info为组件前缀提供 server 名字。在 2026 世代的连接上，client.server_info可能是None（因为身份信息是可选的），这种情况下请自行传入一个Implementation(name=..., version=...)实例。

经典握手（Classic Handshake）：group 如何与 server 建立会话

ClientSessionGroup建立在ClientSession之上，而不是Client之上。因此，每一次connect_to_server都会执行经典的initialize握手，而绝不会发送 docs/protocol-versions.md 中描述的server/discover探测请求。

这一点在源码中非常直观：_establish_session在创建传输和mcp.ClientSession后，直接调用await session.initialize()（session_group.py），把握手结果里的server_info返回给上层用于组件命名。相比之下，基于Client的现代路径会优先尝试更快的server/discover探测协商协议版本（这属于 docs/protocol-versions.md 的主题）。

由于每一个 MCP server 都理解经典initialize握手，group 采取这条路线不会损失任何兼容性；它唯一的代价只是：对于那些本可以通过server/discover走更快路径的 server，group 选择了更老、更慢的方式去访问它。如果你的应用对握手速度敏感、且所有目标 server 都支持新探测协议，可以考虑使用单个Client连接；需要多 server 聚合时，ClientSessionGroup的经典握手则是最稳妥的兼容方案。

小结

• ClientSessionGroup持有多个 server 连接，并把它们的 tools、resources、prompts 各自聚合成一个dict。
• 每个 server 调用一次connect_to_server(params)；它接收传输参数（StdioServerParameters/StreamableHttpParameters/SseServerParameters），而不是Client所接收的 URL 或Transport对象。
• group.call_tool(name, arguments)自动把调用路由到拥有该 tool 的那台 server。
• 名字在整个 group 内必须唯一：两个都叫search的 server 无法直接共存，会抛出MCPError: {'search'} already exist in group tools.
• component_name_hook=重写每一个注册名：dict key 改变，但 wire 上的名字不变。
• connect_with_session添加一个你已持有的 session；disconnect_from_server移除一个 session 及其全部组件。
• group 说的是经典initialize握手，而Client更偏好server/discover快速探测——两者的取舍详见 docs/protocol-versions.md。

想深入验证本文所有结论，可以直接阅读并运行 docs_src/session_groups/ 下的四个教程脚本，以及针对它们的逐条断言测试 tests/docs_src/test_session_groups.py。

• 人工智能
• MCP 服务
• MCP Clients