MCP Python SDKのClientSessionGroupで複数のMCPサーバーと接続する – MNTSQ Techブログ

企業テック

こんにちは、MNTSQ(モンテスキュー)でアルゴリズムエンジニア（AIエンジニア）をしている清水です。

MNTSQのプロダクトをLLMネイティブなプロダクトに進化させるべく、LLMOpsに関する実装が増えてきた今日この頃です。

これらの実装の過程で、複数のMCPサーバーに接続してセッションを管理するにはどのような実装がベストか？という問題にぶつかりました。

自前でラッパークラスを実装するしか方法はないのか、と思っていたのですが、MCP Python SDKにClientSessionGroupというクラスがあることを発見したので、これを使うと良さそうだという結論になりました。

私が調べた限りでは、ClientSessionGroupの使用方法について紹介している記事などは見つけられませんでした。そのため、本記事でMCP Python SDKのClientSessionGroupの仕様や使い方、ClientSessionとの違いなどを整理してまとめてみました。

前提

MCPの実装にはAnthropicのMCP Python SDKのバージョン1.19.0を使用しています。今後のバージョンアップによって仕様が変更する可能性がある点にご留意ください。

github.com

単一MCPサーバーと接続するサンプルコード

単一のMCPサーバーと接続するミニマムなサンプルコードを以下に示します。このサンプルコードは公式GitHubで紹介されているサンプルコードを少し改変したものです。このサンプルコードを複数のMCPサーバーと接続できるように拡張し、比較する形式で説明します。

import asyncio

from mcp import ClientSession
from mcp.client.streamable_http import streamablehttp_client

async def main():
    mcp_server_url = "http://127.0.0.1:8000/echo/mcp"
    async with streamablehttp_client(mcp_server_url) as (read_stream, write_stream, _):
        async with ClientSession(read_stream, write_stream) as session:
            await session.initialize()

            tools = (await session.list_tools()).tools
            print(f"Available tools: {[tool.name for tool in tools]}")

            result = await session.call_tool(
                name="echo", arguments={"message": "Hello, world!"}
            )
            print(f"Call tool result: {result}")

if __name__ == "__main__":
    asyncio.run(main())

streamablehttp_clientを使用してMCPサーバーに接続し、ClientSessionを使用してセッションを作成しています。

MCPサーバーが一つだけならば、このサンプルコードをほぼコピペする形で実装できるのですが、複数MCPサーバーの場合はどのように実装すれば良いでしょうか？

ClientSessionGroupを使って複数のMCPサーバーと接続する

このような用途のためにClientSessionGroupが用意されています。公式のAPIリファレンスでは以下のように紹介されています（筆者による日本語訳）。

複数のMCPサーバーへの接続を管理するためのクライアントクラス。

このクラスは、サーバー接続の管理機能をカプセル化する役割を担う。接続されたすべてのサーバーから提供されるツール、リソース、およびプロンプトを集約する。

先ほどのサンプルコードを複数MCPサーバーへと接続できるように拡張すると以下のようになります。

import asyncio

from mcp import ClientSessionGroup
from mcp.client.session_group import StreamableHttpParameters

async def main():
    mcp_server_urls = [
        "http://127.0.0.1:8000/echo/mcp",
        "http://127.0.0.1:8001/math/mcp",
    ]
    
    async with ClientSessionGroup(
        component_name_hook=lambda name, server_info: f"{server_info.name}.{name}"
    ) as session_group:
        for url in mcp_server_urls:
            
            await session_group.connect_to_server(StreamableHttpParameters(url=url))

        
        tools = session_group.tools.values()
        print(f"Available tools: {[tool.name for tool in tools]}")
        tool_names = session_group.tools.keys()
        print(f"Tool names: {tool_names}")

        
        result = await session_group.call_tool(
            name="EchoServer.echo", args={"message": "Hello, world!"}
        )
        print(f"Call tool result: {result}")

        result = await session_group.call_tool(
            name="MathServer.add_two", args={"n": 10}
        )
        print(f"Call tool result: {result}")

if __name__ == "__main__":
    asyncio.run(main())

単一MCPの場合との差分を一つずつ見ていきます。

1. ClientSessionの代わりにClientSessionGroupを使用してsession_groupを作成

ClientSessionをインスタンス化するときとは異なりread_streamやwrite_streamなどは不要です。空のセッショングループを作成し、後からセッションを追加していく方式であるためです。また、MCPサーバー間でのツール名の衝突を避けるために、component_name_hookを与えることができます。

2. ClientSessionGroup.connect_to_serverを使い、MCPサーバーと接続・セッションを作成

ClientSessionGroup.connect_to_serverにMCPサーバーのURLを渡すだけで、内部でMCPサーバーに接続し、セッションが作成されます。以下にconnect_to_serverメソッド内部の処理を一部抜粋します。

        session_stack = contextlib.AsyncExitStack()
        try:
            
            else:
                client = streamablehttp_client(
                    url=server_params.url,
                    headers=server_params.headers,
                    timeout=server_params.timeout,
                    sse_read_timeout=server_params.sse_read_timeout,
                    terminate_on_close=server_params.terminate_on_close,
                )
                read, write, _ = await session_stack.enter_async_context(client)

            session = await session_stack.enter_async_context(mcp.ClientSession(read, write))
            result = await session.initialize()

実装を確認すると、単一MCPの場合と同様に

  1. streamblehttp_clientでMCPサーバーに接続
  2. ClientSessionでセッションを作成
  3. ClientSession.initializeで初期化

を実行していることがわかります。このように単一MCPの場合と同じように作成したセッションをコンテキストスタックに追加することで、複数セッションを管理しています。

3. ClientSessionGroup.toolsから全てのツールにアクセスする

ClientSessionGroup.toolsで全てのツールにアクセスできます。dictを返すので.values()をつけることで、単一MCPの場合の(await session.list_tools()).toolsと同等のオブジェクトを得ることができます。ただし、以下のような相違点があります。

  1. ClientSessionGroupは、connect_to_serverの実行時に、内部でClientSession.tool_listを呼び出します。ClientSessionGroup.toolsはすでに取得済みのツール群を返すpropertyであり、その場でClientSession.tool_listを呼び出しているわけではありません。
  2. ClientSessionGroup.toolは{”ツール名”: Toolオブジェクト}のdictを返します。この”ツール名”はcomponent_name_hook関数によって付けられた名前です。一方で、Toolオブジェクトのname属性は元のツール名のままであることに注意してください。後にcall_toolを実行するときはcomponent_name_hook関数によって付けられた名前で呼び出す必要があります。よってLLMにToolオブジェクトを与えるときも、component_name_hook関数によって付けられた名前に差し替えたToolオブジェクトを与える必要があります1。

また、promptやresourceについても上記の仕様が当てはまります。

4. ClientSessionGroup.call_toolで各種ツールを実行する

ClientSessionGroup.call_toolはClientSession.call_toolとほぼ同等に使うことができます。以下にClientSessionGroup.call_toolのコードを抜粋します。

    async def call_tool(self, name: str, args: dict[str, Any]) -> types.CallToolResult:
        """Executes a tool given its name and arguments."""
        session = self._tool_to_session[name]
        session_tool_name = self.tools[name].name
        return await session.call_tool(session_tool_name, args)

見ての通り内部の実装は簡素なものです。与えられたツール名がどのセッションに属するツールかを解決する処理が挟まっています。この点を除けば、単にClientSession.call_toolを呼び出す関数と言って良いでしょう。

まとめ

本記事ではMCP Python SDKのClientSessionGroupについて紹介しました。ぜひ開発の参考にしていただけたら幸いです。

MNTSQでは、プロダクトをLLMネイティブに進化させるべく、LLMエージェントを搭載した新機能や、LLMの運用・改善のための基盤（LLMOps）を鋭意開発・構築中です。もしMNTSQの仕事にご興味を持っていただけたら、ぜひお気軽にカジュアル面談でお話ししましょう！

careers.mntsq.co.jp

note.mntsq.co.jp

tech.mntsq.co.jp

この記事を書いた人

tealgreen0503




元の記事を確認する

MCPサーバー開発大全 | 技術評論社 前の記事 Geminiの「Deep Research」、GmailとGoogleドライブにも接続可能に 次の記事

関連記事