10.2 ClientSessionGroup:聚合多个服务端

10.2 ClientSessionGroup:聚合多个服务端

本节摘要:本节讲客户端最有用的进阶抽象——ClientSessionGroup(会话组)。真实场景里,你经常要同时连多个 MCP 服务:一个查数据库、一个读文件、一个调外部 API。会话组让你把这些服务聚合成一个调用面——像调一个服务那样调一群服务,通过命名区分各服务。本节讲透它的注册、统一调用、命名区分机制,用一个编排场景说明它的价值。读完本节,你能用客户端做生产级的多服务编排。

一、为什么需要会话组

先看一个没有会话组的痛点。假设你要同时连三个服务:

# 反模式:手动管多个 Client
async with Client(db_server) as db_client, \
         Client(file_server) as file_client, \
         Client(api_server) as api_client:
    # 调用时要想清楚「用哪个 client」
    db_result = await db_client.call_tool("query", {...})
    file_result = await file_client.read_resource("file://...")
    api_result = await api_client.call_tool("fetch", {...})
    # 手动整合结果
​

这种写法的问题:

• 要手动管多个 client(连接、清理、异常)
• 要记住「哪个工具在哪个 client」
• 没有统一的调用面(每个 client 独立)
• 难以动态增减服务(改了要改一堆代码)

会话组解决这些——把多个服务聚合成一个调用面,自动管理连接,通过命名区分。

二、ClientSessionGroup 的基本用法

会话组让你注册多个服务端,统一调用:

from mcp import Client
from mcp.client.session_group import ClientSessionGroup

async def main():
    group = ClientSessionGroup()

    # 注册多个服务端(用名字区分)
    await group.add_client(Client(db_server), name="db")
    await group.add_client(Client(file_server), name="files")
    await group.add_client(Client(api_server), name="api")

    # 统一调用面
    async with group:
        # 列出所有服务的所有工具
        all_tools = await group.list_tools()
        # 调用时指定来源(或自动路由)
        result = await group.call_tool("query", {...})   # 自动找到 db 的 query
​

关键特性:

• add_client(client, name):注册一个服务端,用名字区分
• 统一调用面:group.call_tool / group.read_resource 等
• 自动路由:工具/资源按注册时路由到对应服务

三、命名区分与调用指定

会话组用名字区分各服务,你可以指定「从哪个服务调」:

async with group:
    # 方式一:自动路由(工具/资源按唯一性找到对应服务)
    result = await group.call_tool("query", {...})

    # 方式二:指定来源(工具名冲突时,用「名字.工具名」)
    result = await group.call_tool("db.query", {...})
    result = await group.call_tool("api.fetch", {...})
​

命名区分解决工具名冲突——如果两个服务都有 query 工具,用 db.query 和 api.query 区分。这是会话组的关键能力,让你能集成工具名重叠的多个服务。

💡 技巧:即使工具名不冲突,也建议用「名字.工具名」的显式形式。这让代码更清晰(一眼看出「这个工具来自哪个服务」),也避免后续加新服务时出现意外冲突。

四、统一调用面的价值

会话组的「统一调用面」带来几个价值:

# 动态增减服务
await group.add_client(Client(new_server), name="new")
# 立即可用
result = await group.call_tool("new.some_tool", {})

await group.remove_client("api")    # 移除 api 服务
# api 的工具不再可用,其他不受影响
​

五、一个编排场景

用一个真实场景说明会话组的价值——「用数据库服务 + 文件服务 + API 服务协作完成一个任务」:

async def analyze_user(user_id: str):
    async with group:
        # 1. 从数据库服务查用户
        user = await group.call_tool("db.get_user", {"id": user_id})
        # user.structured_content = {"name": "...", "region": "..."}

        # 2. 从文件服务读该地区的配置
        region = user.structured_content["region"]
        config = await group.read_resource(f"files://config/{region}")

        # 3. 用 API 服务调外部接口,带上配置
        result = await group.call_tool("api.call", {
            "endpoint": "/analyze",
            "data": {"user": user.structured_content, "config": config}
        })

        return result
​

注意:三个服务的工具/资源,通过 group 统一调用,代码像调一个服务一样简洁。没有会话组,这段代码要手动管三个 client、记清来源、整合结果——复杂得多。

六、会话组与 Host 的关系

会话组是「自研 Host」的核心抽象。回顾第 2.1 节的「一宿主多客户端」模型——一个宿主连多个服务,每个跑一个客户端。会话组正是这个模型的实现:

你自研的 Host
   │
   ├─ ClientSessionGroup
   │   ├─ db 客户端 ──► 数据库服务
   │   ├─ files 客户端 ──► 文件服务
   │   └─ api 客户端 ──► 外部 API 服务
   │
   └─ 把多个服务的能力聚合成一个对话面
       (模型看到统一工具清单,不知道背后是三个服务)
​

如果你要做自己的 AI 应用(集成 LLM + 多个 MCP 服务),会话组是编排层的核心。

⚠️ 注意:会话组替你管理「调用哪个服务」,但不替你管理「模型如何用这些工具」。模型推理、工具调用决策仍需你集成(用 OpenAI、Anthropic 等的 SDK)。会话组解决「多服务消费」,不解决「LLM 集成」。

七、会话组的进阶能力预告

会话组还有一些进阶能力,本章后续会讲:

这些能力让会话组从「简单聚合」升级为「生产级编排系统」。

本节要点回顾

1. ClientSessionGroup 把多个服务聚合成一个调用面,解决手动管多 client 的痛点。
2. add_client(client, name) 注册服务,用名字区分。
3. 统一调用面:group.call_tool / group.read_resource,自动路由或显式指定。
4. 「名字.工具名」解决工具名冲突,建议即使不冲突也用显式形式。
5. 价值:简化调用、自动管理连接、动态增减、名字空间隔离。
6. 编排场景:多服务协作完成任务,代码像调一个服务一样简洁。
7. 会话组是自研 Host 的核心抽象,解决「多服务消费」(不解决 LLM 集成)。

会话组清楚了,下一节讲跨服务缓存——避免重复调用昂贵工具。