はじめに
最近、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 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月）ではこのアプローチが有効です。
マルチエージェントやセッション管理に課題を感じている方は、ぜひ本構成を参考にしてみてください。