はじめに

最近、Google が Agent Development Kit（ADK）をリリースしました。これにより、手軽に AI エージェントを開発・テストできるようになりました。

ところが、ADK に付属するテスト用「ADK Web Server」では、SSE 対応の MCP サーバーをツールとして指定した際に、エージェントが正しく動作しないという問題が発生しました。

いくら調べても明確な解決方法が見つからなかったため、試行錯誤を重ね、なんとか動作させる方法を見つけました。本記事では、ADK Web Server と MCP サーバーを組み合わせて AI エージェントを動かす方法をまとめます。
（※2025/4/15 時点の ADK バージョンは 0.1.0 です。将来的にバージョンアップでこの対応が不要になる可能性もあります）

背景

ADK は、Google が提供する AI エージェント開発用のフレームワークで、CLI や Web Server を通じて簡単にエージェントの開発・テストが行えます。
しかし、ADK Web Server と SSE 対応 MCP サーバーを組み合わせた場合、セッション管理の問題などにより、エージェントがうまく認識されないことがありました。
本記事では、ADK Web Server + MCP サーバー + AI エージェントの構成で動作させる手順を紹介します。

使用したリポジトリ

本記事は以下の GitHub リポジトリをベースにしています。

adk-mcp-starter

このリポジトリには、ADK Web Server・MCP サーバー・AI エージェントを連携させるためのサンプルコードが含まれており、
自然言語リクエストを MCP サーバー経由で処理し、結果を返す一連のフローを確認できます。

使用技術

Python 3.12+
Google Agent Development Kit (ADK) v0.1.0
Model Context Protocol (MCP) v1.6.0

セットアップ手順

1. リポジトリのクローン

git clone https://github.com/Kewton/adk-mcp-starter.git
cd adk-mcp-starter

※ README 上では cd mcp-gadk-sample となっていますが、クローン後のフォルダ名に応じて調整してください。

2. MCP サーバーのセットアップ

cd sample-mcp-server
python3 -m venv venv
source venv/bin/activate
pip install -r requirements.txt

3. ADK エージェントのセットアップ

cd ../sample-adk-agent
python3 -m venv venv
source venv/bin/activate
pip install -r requirements.txt

4. 実行と動作確認（ADK Web Server 経由）

(1) `.env` ファイルの作成

sample-mcp-server/.env

HOST=127.0.0.1
PORT=8001

sample-adk-agent/.env

SSE_SERVER_PARAMS_URL=http://localhost:8001/sse
MODEL=gemini-2.0-flash
GOOGLE_API_KEY=<your api key>

(2) MCP サーバーの起動

cd sample-mcp-server
python main.py

(3) ADK Web Server の起動

cd ../sample-adk-agent
adk web

起動後、「For local testing, access at http://...」と表示される URL にアクセスすると、ADK Web Server の GUI が起動します。

(4) AI エージェントに指示を送信

画面左上のドロップダウンから "sampleagent" を選択
右下の入力フィールドから自然言語で指示を送信 → MCP サーバー経由で実行され、結果が表示されます

やったこと・ハマりポイント

ADK Web Server 経由で AI エージェントを動かすためには、root_agent という名前で google.adk.agents.llm_agent.LlmAgent のインスタンスを定義する必要があります。

ただし、LlmAgent の初期化時に MCP サーバーのツール情報やセッション情報を非同期で取得する設計となっており、この処理が ADK Web Server からはうまく動作しません。

そこで、LlmAgent を継承した McpAgentToolAgent クラスを新たに実装し、以下の 2 点をカスタム処理として追加しました：

MCP サーバーから動的にツール情報を取得して上書き
ツール使用後にセッション情報をクリーンにクローズ

🔧 `McpAgentToolAgent` 実装のポイント

./sample-adk-agent/common/base.py

from google.adk.agents.llm_agent import (
    LlmAgent,
    InvocationContext,
    Event,
)
from google.adk.tools.mcp_tool.mcp_toolset import MCPToolset, SseServerParams
from typing import AsyncGenerator, Optional
from typing_extensions import override
from contextlib import AsyncExitStack


class McpAgentToolAgent(LlmAgent):

    # インスタンスごとにMCPサーバーのURLを保持する変数
    mcp_server_url: str

    @override
    async def _run_async_impl(
        self, ctx: InvocationContext
    ) -> AsyncGenerator[Event, None]:
        """
        MCPサーバーから動的にツールを取得し、コアロジックを実行後、リソースをクリーンアップします。
        """
        print(f"Agent '{self.name}': Entering custom _run_async_impl.")

        original_tools = list(self.tools)  # 元のツールをコピーして保持 (変更される可能性があるため)
        exit_stack: Optional[AsyncExitStack] = None  # exit_stackを初期化

        try:
            # 1. MCPサーバーからツールとExitStackを取得
            print(f"Agent '{self.name}': Fetching tools from MCP server ({self.mcp_server_url})...")

            # MCPToolset.from_server は接続を確立し、クリーンアップ用の exit_stack を返す
            tools, exit_stack = await MCPToolset.from_server(
                connection_params=SseServerParams(url=self.mcp_server_url)
            )
            self.tools = tools  # この実行のためにツールを再設定
            tool_names = [getattr(t, 'name', str(t)) for t in self.tools]  # 安全に名前を取得
            print(f"Agent '{self.name}': Tools reconfigured to: {tool_names}")

            # 2. 親クラスのコアロジックを実行 (新しいツールが使われる)
            print(f"Agent '{self.name}': Calling super()._run_async_impl to process request...")
            async for event in super()._run_async_impl(ctx):
                yield event
            print(f"Agent '{self.name}': Finished processing request in super()._run_async_impl.")

        except Exception as e:
            print(f"Agent '{self.name}': Error during execution: {e}", exc_info=True)
            # エラー発生時は元のツールに戻す (オプション)
            self.tools = original_tools
            print(f"Agent '{self.name}': Reverted to original tools due to error.")
            # エラーを再raiseするか、エラーイベントをyieldするかなどを検討
            # raise e # エラーを呼び出し元に伝える場合
        finally:
            # 3. 取得したexit_stackをクローズしてリソースを解放
            print(f"Agent '{self.name}': Entering finally block for cleanup.")
            if exit_stack:
                print(f"Agent '{self.name}': Closing MCP toolset exit_stack...")
                try:
                    await exit_stack.aclose()
                    print(f"Agent '{self.name}': MCP toolset exit_stack closed successfully.")
                except Exception as close_exc:
                    # クリーンアップ中のエラーもログに残す
                    print(f"Agent '{self.name}': Error closing exit_stack: {close_exc}", exc_info=True)
            else:
                print(f"Agent '{self.name}': No exit_stack was acquired, skipping close.")

            print(f"Agent '{self.name}': Exiting custom _run_async_impl.")

`McpAgentToolAgent` の使用例

./sample-adk-agent/sampleagent/agent.py

from common.base import McpAgentToolAgent
import os


SSE_SERVER_PARAMS_URL = os.getenv('SSE_SERVER_PARAMS_URL', "http://localhost:8001/sse")
MODEL = os.getenv('MODEL', "gemini-1.5-flash")  # Adjust this to your desired model


root_agent = McpAgentToolAgent(
    model=MODEL,  # Adjust model name if needed based on availability
    name='html_parser',
    instruction='Fetch HTML from a URL, parse it, and convert the content to Markdown.',
    tools=[],  # Provide the MCP tools to the ADK agent
    mcp_server_url=SSE_SERVER_PARAMS_URL,  # URL to the MCP server
)

まとめ

本記事では、Google の ADK Web Server と SSE MCP サーバーを連携させて、AI エージェントを正しく動作させる方法を紹介しました。

GUI で直感的にテストできる ADK Web Server に、柔軟なツール実行を担う MCP サーバーを組み合わせることで、よりリアルなエージェントの開発・評価が可能になります。

将来的に公式でのサポートや仕様変更が入る可能性もありますが、現時点（2025年4月）ではこのアプローチが有効です。
マルチエージェントやセッション管理に課題を感じている方は、ぜひ本構成を参考にしてみてください。

参考リンク