工作階段群組

一個 Client 只連到一台伺服器。實際的應用程式往往需要好幾台（搜尋伺服器、資料庫伺服器、內部 API），結果得替每一台各自管理一條連線和一份工具清單。

ClientSessionGroup 是單一物件，裡面握有多條連線，並把它們公開的所有東西合併成一個統一的檢視。

兩台伺服器

先從兩台普通的伺服器開始。它們彼此毫無關係，所以很自然地都把自己的工具取名為 search：

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"

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}."

一個群組

建立一個 ClientSessionGroup，然後對每台伺服器各呼叫一次 connect_to_server：

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())

• connect_to_server 接受的是傳輸參數，不是伺服器物件：用 StdioServerParameters（來自 mcp）啟動子處理程序，或用 StreamableHttpParameters / SseServerParameters（來自 mcp.client.session_group）連到已經在某個 URL 上監聽的伺服器。
• group.tools 是一個 dict[str, Tool]，收集所有已連線伺服器的工具。group.resources 和 group.prompts 的結構相同。
• group.call_tool(name, arguments) 會查詢名稱、找出擁有它的工作階段（session），再把呼叫轉送過去。你永遠不需要指明是哪台伺服器。

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

component_name_hook

這個問題要在群組這邊解決，而不是在伺服器端。傳入一個接收 (name, server_info) 的函式，群組會對它登記的每個名稱執行這個函式：

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)) 現在兩個都會顯示：

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

• 鍵是你自己決定的。by_server 用 server_info.name 組出來，也就是每個 MCPServer(...) 建構時傳入的名稱。
• 裡面的 Tool 完全沒動：group.tools["Web.search"].name 仍然是 "search"，而這也是 call_tool 放上線路的名稱。前綴永遠不會離開你的處理程序。
• 不只工具如此。圖書館的 hours 資源登記為 Library.hours。

新增與移除伺服器

connect_to_server 會回傳它開啟的 ClientSession。如果之後可能想拿掉那台伺服器，就把它留著：await group.disconnect_from_server(session) 會把它的工具、資源和提示詞從群組中移除。

如果手上已經有一個連線中的 ClientSession（Client.session 就是一個），改把它交給 await group.connect_with_session(server_info, session)，不必另開新的傳輸。它彙整的方式相同。群組永遠不會關閉不是它自己開啟的工作階段。server_info 用來替伺服器命名，供元件前綴使用；在 2026 世代的連線上，client.server_info 可能是 None（身分是選填的），這種情況下就傳入你自己的 Implementation(name=..., version=...)。

傳統的交握

ClientSessionGroup 建立在 ClientSession 之上，而不是 Client。每次 connect_to_server 都會執行傳統的 initialize 交握，從不送出 協定版本 裡描述的 server/discover 探測。每台 MCP 伺服器都懂這套交握，所以這不會讓你犧牲任何相容性；只是代表群組面對一台本來能做得更好的伺服器時，走的是比較舊、比較慢的路徑。

重點回顧

• ClientSessionGroup 握有多條伺服器連線，並把它們的工具、資源和提示詞各自合併成一個 dict。
• 每台伺服器呼叫一次 connect_to_server(params)。它接受傳輸參數，絕不是 Client 接受的伺服器物件或 URL。
• group.call_tool(name, arguments) 會替你轉送到擁有該工具的伺服器。
• 名稱在整個群組內必須唯一；兩台都有 search 工具的伺服器無法原樣共存。
• component_name_hook= 會改寫每個登記的名稱。dict 的鍵會變，線路上的名稱不變。
• connect_with_session 加入你已經握有的工作階段；disconnect_from_server 移除一個。

群組使用的交握（以及 Client 偏好的那套更快的交握）是 協定版本 的主題。