python-sdk 会话组（ClientSessionGroup）完全指南：用一个对象聚合管理多个 MCP 服务器的工具、资源与提示词

人工智能MCP 服务MCP Clients【免费下载链接】python-sdkThe official Python SDK for Model Context Protocol servers and clients项目地址https://gitcode.com/gh_mirrors/pythonsd/python-sdk点击查看免费下载导读在 MCP 的官方 Python SDKpython-sdk中一个Client只能连接一个服务器而真实应用往往同时需要搜索服务器、数据库服务器、内部 API 等多个服务。本文围绕官方文档 docs/client/session-groups.md 展开深入讲解ClientSessionGroup—— 这个一个对象持有多个连接、并把所有服务器暴露的工具tools、资源resources和提示词prompts合并成统一视图的聚合抽象。读完本文你将掌握如何用connect_to_server一次连接多个服务器、如何通过component_name_hook解决跨服务器命名冲突、如何动态增删服务器以及为什么组走的是经典initialize握手而非server/discover探测。文中所有结论都有源码与测试用例印证。背景为什么需要会话组单个Client的生命周期内只维护一条与服务器的连接。当应用需要同时调用多个 MCP 服务器时如果逐个创建Client你就不得不为每台服务器各自维护一份连接对象和一份工具清单并在自己的业务代码里手工路由每次调用——这正是官方文档描述的“juggling a connection and a tool list for each”的窘境。ClientSessionGroup正是为这个问题而生它是一个对象内部持有许多连接并把它们暴露的一切组件工具、资源、提示词合并成单一视图。调用方只面对一组名字不需要关心名字属于哪台服务器。起点两个互不相干的服务器先看最典型的冲突场景。假设你有两台普通服务器它们之间没有任何关系因此都自然而然地把自己唯一的工具命名为searchlibrary_server.py来自 docs_src/session_groups/tutorial001.pyfrom mcp.server import MCPServer mcp MCPServer(Library) mcp.tool() def search(query: str) - str: Search the library catalog. return f3 books match {query!r}. mcp.resource(library://hours) def hours() - str: When the library is open. return Mon-Fri 09:00-17:00web_server.py来自 docs_src/session_groups/tutorial002.pyfrom mcp.server import MCPServer mcp MCPServer(Web) mcp.tool() def search(query: str) - str: Search the web. return f12 pages match {query!r}.注意MCPServer(Library)与MCPServer(Web)中的名字它将成为server_info.name也是后面component_name_hook做前缀的原料。library_server.py除了工具外还暴露了一个名为hours的资源稍后我们会看到它在聚合视图里的表现。创建一个组connect_to_server把两台服务器接入同一个组核心代码在 docs_src/session_groups/tutorial003.pyimport asyncio from mcp import ClientSessionGroup, StdioServerParameters async def main() - None: library StdioServerParameters(commanduv, args[run, mcp, run, library_server.py]) web StdioServerParameters(commanduv, 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())这里有三个要点connect_to_server接收的是传输参数transport parameters而不是服务器对象也不是Client所接收的 URL 或Transport。官方文档明确区分了三种参数类型StdioServerParameters从mcp导入——用于以子进程方式启动服务器如本例的uv run mcp run ...StreamableHttpParameters/SseServerParameters从mcp.client.session_group导入——用于连接已经在某个 URL 上监听的服务器。group.tools是一个dict[str, Tool]汇总了所有已连接服务器的工具group.resources和group.prompts形状相同。group.call_tool(name, arguments)负责路由它按名字查找、找到拥有该工具的会话并把调用转发过去调用方永远不用指定“发给哪台服务器”。源码视角聚合是如何发生的从源码看connect_to_server的链路非常清晰见 src/mcp/client/session_group.py_establish_session根据参数类型分别进入stdio_client、sse_client或streamable_http_client然后包装出一个mcp.ClientSession并执行session.initialize()握手把initialize返回的server_info和 session 一起交出去src/mcp/client/session_group.py。connect_with_session调用_aggregate_components对会话依次执行list_prompts、list_resources、list_tools把返回的组件写进三个临时字典src/mcp/client/session_group.py。之所以用临时字典是因为实现上不允许“中途失败却留下半截数据”只有三份清单都取回、且全部通过重名校验后才会一次性更新self._prompts、self._resources、self._tools以及反向索引self._tool_to_session用于把工具名映射回所属 session。每接入一台服务器组件名都会记录在该 session 的_ComponentNames反向索引中供日后disconnect_from_server精确清理。冲突同名工具无法共存把client.py放在两台服务器旁边直接运行第二次connect_to_server会拒绝并抛出异常官方文档的!!! check警示块原文mcp.shared.exceptions.MCPError: {search} already exist in group tools.这是MCPError并且是在第二个服务器的任何组件被登记之前就抛出的——异常发生时第一个服务器的search依然保留在组里第二个服务器的任何东西都不会残留。测试 tests/docs_src/test_session_groups.py 精确复现了这一点断言报错文案、错误码为INVALID_PARAMS并且sorted(group.tools) [search]第二个服务器的组件确实没有被写入。从源码看重名检测发生在_aggregate_components的末尾src/mcp/client/session_group.py对 prompts、resources、tools 三份临时字典逐一与组内已有键做集合交集运算prompts_temp.keys() self._prompts.keys()一旦命中就抛出MCPError。这意味着名字必须在整个组内唯一——不只是工具资源与提示词同样受此约束两台你不控制的服务器迟早会发生命名冲突这不是偶然而是必然。解决方案component_name_hook冲突要在组这一层修复而不是去改服务器。向ClientSessionGroup传入一个签名为(name, server_info)的函数组在登记每一个名字时都会执行它见 docs_src/session_groups/tutorial004.pyimport 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(commanduv, args[run, mcp, run, library_server.py]) web StdioServerParameters(commanduv, args[run, mcp, run, web_server.py]) async with ClientSessionGroup(component_name_hookby_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))将同时显示两台服务器的工具[Library.search, Web.search]关于这个 hook官方文档给出了三个关键结论测试也逐一验证字典键由你决定。by_server用server_info.name拼出键名——也就是每个MCPServer(...)构造时传入的名字。测试test_the_hook_is_a_plain_function_of_name_and_server_info直接断言by_server(search, Implementation(nameWeb, version1.0.0)) Web.search见 tests/docs_src/test_session_groups.py。字典里的Tool对象原封不动group.tools[Web.search].name仍然是search而call_tool在线路上发送的也正是这个原始名——前缀永远不会离开你的进程tests/docs_src/test_session_groups.py 专门断言了“键被加前缀、线上名字不变”。不只是工具。library 服务器的hours资源同样被注册为Library.hours测试中断言sorted(group.resources) [Library.hours]。另外官方文档用一个!!! tip特别提醒hook 会对每台服务器的每个名字都执行而不是只在发生冲突时才执行——不存在“仅在冲突时加前缀”的模式。所以请选定一种命名方案让它统一作用于所有名字。源码视角hook 的调用位置实现非常直白src/mcp/client/session_group.pydef _component_name(self, name: str, server_info: types.Implementation) - str: if self._component_name_hook: return self._component_name_hook(name, server_info) return name_aggregate_components中list_prompts、list_resources、list_tools返回的每个组件名都会先经过self._component_name(...)再写入临时字典若未配置 hook则原样返回。同时注意反向索引_ComponentNames里记录的也是经过 hook 后的名字这保证了 disconnect 时能精确删除带前缀的组件。调用路由call_tool如何找到正确的服务器call_tool的实现src/mcp/client/session_group.py先通过self._tool_to_session[name]找到拥有该工具的 session再取self.tools[name].name作为真正发往线路上名字session self._tool_to_session[name] session_tool_name self.tools[name].name return await session.call_tool(session_tool_name, argumentsarguments, ...)也就是说你调用group.call_tool(Web.search, ...)底层实际调用的是 web 服务器的session.call_tool(search, ...)。测试test_call_tool_routes_to_the_owning_server验证了这一点——group.call_tool(Web.search, ...)拿到的是 web 服务器的结果group.call_tool(Library.search, ...)拿到的是 library 服务器的结果tests/docs_src/test_session_groups.py。此外call_tool还透传了read_timeout_seconds、progress_callback、input_responses、request_state、meta、allow_input_required等参数并支持返回InputRequiredResult见单元测试 tests/client/test_session_group.py。动态增删服务器disconnect_from_server与connect_with_session会话组不是静态的官方文档专门讲解了两种动态操作connect_to_server会返回它打开的ClientSession。如果你想在将来把某台服务器移出组请保留这个返回值然后调用await group.disconnect_from_server(session)这会从组的工具、资源和提示词字典中移除该服务器的一切组件。从源码看src/mcp/client/session_group.py它依赖前面提到的_ComponentNames反向索引逐个删除该会话登记过的 prompts、resources、tools以及_tool_to_session中的对应条目并关闭该会话专属的退出栈。如果传入的 session 既不在组件索引里、也不在退出栈索引里会抛出MCPError消息为 “Provided session is not managed or already disconnected.”测试test_client_session_group_disconnect_from_server与test_client_session_group_disconnect_non_existent_server分别覆盖了正常移除与误删未管理会话两种情况。docs 测试test_disconnect_removes_every_component_of_that_server还验证了断开后sorted(group.tools) [Library.search]、sorted(group.resources) [Library.hours]即第二个服务器的组件被完整移除tests/docs_src/test_session_groups.py。如果你已经持有一条已连接的ClientSession例如Client.session就是一条就不要新开传输而是交给await group.connect_with_session(server_info, session)它走完全相同的聚合路径_aggregate_components并且组永远不会关闭它自己没有打开的会话——这保证了所有权的清晰边界。server_info用来为组件前缀命名在 2026 时代的连接上client.server_info可能为None身份标识是可选的此时你需要自己传一个Implementation(name..., version...)。测试 tests/docs_src/test_session_groups.py 中的_server_info辅助函数正是为此设计断言client.server_info非空后再传入。另外值得一提的实现细节ClientSessionGroup.__aexit__会用anyio.create_task_group()并发关闭所有会话的退出栈src/mcp/client/session_group.py因此async with ClientSessionGroup() as group:退出时能干净地批量回收全部连接资源。经典握手为什么组不走server/discover官方文档在“The classic handshake”一节点明了ClientSessionGroup与Client的一个根本差异ClientSessionGroup构建在ClientSession之上而不是Client之上。每次connect_to_server执行的都是经典的initialize握手。它永远不会发送docs/protocol-versions.md 中描述的server/discover探测。这一点可以在源码中得到印证_establish_session中创建ClientSession后只调用了await session.initialize()全程没有server/discoversrc/mcp/client/session_group.py。而Client一侧则实现了server/discover探测与回退逻辑从 src/mcp/client/session.py 可以看到initialize之后会尝试探测并把结果解析进self._discover_server_infoserver_info属性优先返回探测结果、否则退回initialize的结果src/mcp/client/session.py。这意味着什么根据协议版本说明docs/protocol-versions.md2026-07-28 之前的服务器用initialize握手建立连接而 2026-07-28 时代的服务器则改用server/discover单次探测。会话组走经典握手兼容性无损——所有 MCP 服务器都理解initialize握手所以不会失去任何兼容性代价是路径更旧、更慢——对于一个本可以做得更好的服务器组选择的是更保守的握手路径。如果你的应用需要“以最新协议最快建连”请使用Client默认modeauto先探测、失败再回退而把ClientSessionGroup留给“多服务器聚合 保守兼容”的场景。小结ClientSessionGroup持有多个服务器连接把它们的工具、资源和提示词各自合并为一个dict。每台服务器调用一次connect_to_server(params)它接收的是传输参数StdioServerParameters/StreamableHttpParameters/SseServerParameters而不是Client接收的 URL 或Transport对象。group.call_tool(name, arguments)自动把调用路由到拥有该名字的服务器调用方无需指定服务器。名字必须在整个组内唯一两个都叫search的服务器默认无法共存会抛MCPError。component_name_hook重写每个登记的名字字典键变了但发往线路上名字不变前缀只存在于你的进程内。connect_with_session把一条你已持有的会话加入聚合disconnect_from_server把一条会话及其全部组件移出组从不关闭它未打开的会话。组讲的是经典initialize握手Client则优先server/discover该话题的完整讨论见 docs/protocol-versions.md。延伸阅读仓库内官方文档原文docs/client/session-groups.md核心实现src/mcp/client/session_group.py配套教程源码docs_src/session_groups/行为测试tests/docs_src/test_session_groups.py 与 tests/client/test_session_group.py相关协议背景docs/protocol-versions.md赞分享人工智能MCP 服务MCP Clients【免费下载链接】python-sdkThe official Python SDK for Model Context Protocol servers and clients项目地址https://gitcode.com/gh_mirrors/pythonsd/python-sdk点击查看免费下载相关推荐python-sdk 会话组ClientSessionGroup实战一个对象聚合多个 MCP 服务器的工具、资源与提示词python sdk 会话组ClientSessionGroup实战一个对象聚合多个 MCP 服务器的工具、资源与提示词 导读 在真实应用中一个 MCP人工智能MCP 服务MCP Clientspython-sdk 多服务器聚合指南用 ClientSessionGroup 统一管理多个 MCP 连接python sdk 多服务器聚合指南用 ClientSessionGroup 统一管理多个 MCP 连接 ClientSessionGroup 是 pyth人工智能MCP 服务MCP Clients5分钟终极指南如何用Live Server告别手动刷新提升前端开发效率300%5分钟终极指南如何用Live Server告别手动刷新提升前端开发效率300% 还在为每次修改代码后都要手动刷新浏览器而烦恼吗 今天我要为你介绍一款能人工智能MCP 服务MCP Clients上一篇青龙面板参数失效3步彻底解决配置不生效问题下一篇Unity ML-Agents 入门指南从零开始训练3D平衡球AI创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考