人工智能MCP 服务MCP Clients【免费下载链接】python-sdkThe official Python SDK for Model Context Protocol servers and clients项目地址https://gitcode.com/gh_mirrors/pythonsd/python-sdk点击查看免费下载导读单个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.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:00第二个是“网页搜索”服务 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}.注意观察两点两个 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.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())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参数模型关键字段说明StdioServerParameterscommand、args、env等启动子进程的 stdio 传输参数从mcp顶层导入SseServerParametersurl、headers、timeout5.0、sse_read_timeout300.0传统 SSE 传输timeout为常规操作 HTTP 超时秒sse_read_timeout为 SSE 读超时秒headers可携带认证等请求头StreamableHttpParametersurl、headers、timeout30.0、sse_read_timeout300.0、terminate_on_closeTrue现代 Streamable HTTP 传输terminate_on_close控制传输关闭时是否同时关闭客户端会话此外connect_to_server还接受第二个可选参数session_params: ClientSessionParameterssession_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.pygroup 先把新 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.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))现在能同时看到两个 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仍然是searchcall_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.pysession 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。动态添加与移除 Serverconnect_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.hourstests/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 Handshakegroup 如何与 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走更快路径的 servergroup 选择了更老、更慢的方式去访问它。如果你的应用对握手速度敏感、且所有目标 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添加一个你已持有的 sessiondisconnect_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【免费下载链接】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 连接 ClientSessionGroup 是 pyth人工智能MCP 服务MCP Clientspython-sdk 中的 ClientSessionGroup用单一视图聚合管理多个 MCP 服务器连接python sdk 中的 ClientSessionGroup用单一视图聚合管理多个 MCP 服务器连接 ClientSessionGroup 是 Mode人工智能MCP 服务MCP Clients终极Vegeta HTTP负载测试工具指南从入门到性能优化全攻略终极Vegeta HTTP负载测试工具指南从入门到性能优化全攻略 Vegeta是一款功能强大的HTTP负载测试工具和库能够帮助开发者模拟高并发场景测试We人工智能MCP 服务MCP Clients创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考