セッショングループ
Client は 1 つのサーバーに接続します。実際のアプリケーションでは複数のサーバー(検索サーバー、データベースサーバー、社内 API など)を使いたいことが多く、結局それぞれの接続とツール一覧を個別に管理することになります。
ClientSessionGroup は、多数の接続を保持し、それらが公開するものすべてを 1 つのビューにまとめる単一のオブジェクトです。
2 つのサーバー
まず、ごく普通のサーバーを 2 つ用意します。互いに何の関係もないので、どちらも自然とツールに 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}."
1 つのグループ
ClientSessionGroup を作成し、サーバーごとに connect_to_server を 1 回ずつ呼び出します。
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から)、すでに URL で待ち受けているサーバーならStreamableHttpParametersまたはSseServerParameters(mcp.client.session_groupから)です。group.toolsは、接続しているすべてのサーバーのツールを集めたdict[str, Tool]です。group.resourcesとgroup.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) を受け取る関数を渡すと、グループは登録するすべての名前に対してその関数を実行します。
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として登録されます。
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_info が None になることがある(識別情報は任意です)ため、その場合は自分で Implementation(name=..., version=...) を渡してください。
従来のハンドシェイク
ClientSessionGroup は Client ではなく 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 が優先するより高速なハンドシェイク)について詳しくは、プロトコルバージョンを参照してください。