コンテンツにスキップ

セッショングループ

機械翻訳

このページは英語版ドキュメントから自動翻訳されたものであり、正式な版は英語版のページです。不自然な箇所があれば、翻訳についてで報告の方法を説明しています。

Client は 1 つのサーバーに接続します。実際のアプリケーションでは複数のサーバー(検索サーバー、データベースサーバー、社内 API など)を使いたいことが多く、結局それぞれの接続とツール一覧を個別に管理することになります。

ClientSessionGroup は、多数の接続を保持し、それらが公開するものすべてを 1 つのビューにまとめる単一のオブジェクトです。

2 つのサーバー

まず、ごく普通のサーバーを 2 つ用意します。互いに何の関係もないので、どちらも自然とツールに search という名前を付けています。

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

1 つのグループ

ClientSessionGroup を作成し、サーバーごとに connect_to_server を 1 回ずつ呼び出します。

client.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())
  • connect_to_server はサーバーオブジェクトではなく、トランスポートのパラメーターを受け取ります。サブプロセスを起動するなら StdioServerParametersmcp から)、すでに URL で待ち受けているサーバーなら StreamableHttpParameters または SseServerParametersmcp.client.session_group から)です。
  • group.tools は、接続しているすべてのサーバーのツールを集めた dict[str, Tool] です。group.resourcesgroup.prompts も同じ形です。
  • group.call_tool(name, arguments) は名前を引き、それを所有するセッションを見つけて呼び出しを転送します。どのサーバーかを指定する必要はありません。

Check

client.py を 2 つのサーバーと同じ場所に置いて実行してください。2 回目の connect_to_server は拒否されます。

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

これは MCPError で、2 つ目のサーバーの何かが登録される前に送出されます。名前はグループ全体で一意でなければならず、自分で管理していない 2 つのサーバーはいずれ衝突します。

component_name_hook

これはサーバー側ではなく、グループ側で解決します。(name, server_info) を受け取る関数を渡すと、グループは登録するすべての名前に対してその関数を実行します。

client.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)) には両方が表示されます。

['Library.search', 'Web.search']
  • キーは自分で決めたものです。by_serverserver_info.name、つまり各 MCPServer(...) の構築時に渡された名前からキーを組み立てました。
  • 中の Tool は変更されていません。group.tools["Web.search"].name は依然として "search" であり、call_tool が通信路に載せるのはこの名前です。プレフィックスがプロセスの外に出ることはありません。
  • ツールだけではありません。ライブラリの hours リソースは Library.hours として登録されます。

Tip

フックは衝突したものだけでなく、すべてのサーバーのすべての名前に対して実行されます。衝突時だけプレフィックスを付けるモードはありません。1 つの方式を決めて、全体に適用してください。

サーバーの追加と削除

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_infoNone になることがある(識別情報は任意です)ため、その場合は自分で Implementation(name=..., version=...) を渡してください。

従来のハンドシェイク

ClientSessionGroupClient ではなく ClientSession の上に構築されています。connect_to_server を呼ぶたびに従来の initialize ハンドシェイクが実行されます。プロトコルバージョンで説明している server/discover プローブを送ることはありません。このハンドシェイクはすべての MCP サーバーが理解するので、互換性が失われることは一切ありません。ただ、もっと良い方法に対応しているサーバーに対しても、グループは古くて遅い経路を取るというだけです。

まとめ

  • ClientSessionGroup は多数のサーバー接続を保持し、それらのツール、リソース、プロンプトをそれぞれ 1 つの dict にまとめます。
  • サーバーごとに connect_to_server(params) を呼びます。受け取るのはトランスポートのパラメーターであり、Client が受け取るサーバーオブジェクトや URL ではありません。
  • group.call_tool(name, arguments) は、所有するサーバーへのルーティングを代わりに行います。
  • 名前はグループ全体で一意でなければなりません。search ツールを持つ 2 つのサーバーは、そのままでは共存できません。
  • component_name_hook= は登録されるすべての名前を書き換えます。dict のキーは変わりますが、実際に送信される名前は変わりません。
  • connect_with_session はすでに持っているセッションを追加し、disconnect_from_server はセッションを削除します。

グループが使うハンドシェイク(と、Client が優先するより高速なハンドシェイク)について詳しくは、プロトコルバージョンを参照してください。