この Codelab について
1. 概要
AI エージェントの人気は急速に高まっており、自律的に動作し、学習し、環境とやり取りして目標を達成する能力により、タスクの自動化と意思決定に革命をもたらしています。
では、エージェントを構築する具体的な方法を見てみましょう。この Codelab では、さまざまな国の通貨を変換できる通貨エージェントを構築する方法について説明します。最新のテクノロジーを説明することで、インターネットでよく見かける頭字語（MCP、ADK、A2A）を理解できるようにすることを目的としています。
Model Context Protocol（MCP）
Model Context Protocol（MCP）は、アプリケーションが LLM にコンテキストを提供する方法を標準化するオープン プロトコルです。MCP は、AI モデルをリソース、プロンプト、ツールに接続するための標準化された方法を提供します。
エージェント開発キット（ADK）
エージェント開発キット（ADK）は、AI エージェントの開発とデプロイのための柔軟なオーケストレーション フレームワークです。ADK はモデルに依存せず、デプロイに依存せず、他のフレームワークとの互換性を備えて構築されています。ADK は、エージェント開発をソフトウェア開発のように感じて、デベロッパーが簡単なタスクから複雑なワークフローまで、エージェント アーキテクチャを簡単に作成、デプロイ、オーケストレートできるように設計されています。
Agent2Agent（A2A）プロトコル
Agent2Agent（A2A）プロトコルは、AI エージェント間のシームレスな通信とコラボレーションを可能にするために設計されたオープン スタンダードです。MCP が LLM にデータとツールへのアクセス権を付与するための標準化された方法を提供するように、A2A はエージェントが他のエージェントとやり取りするための標準化された方法を提供します。エージェントがさまざまなフレームワークを使用してさまざまなベンダーによって構築されている世界において、A2A は共通言語を提供し、サイロを解体して相互運用性を促進します。
学習内容
- ローカル MCP サーバーを作成する方法
- MCP サーバーを Cloud Run にデプロイする
- MCP ツールを使用するエージェント開発キットを使用してエージェントを構築する方法
- ADK エージェントを A2A サーバーとして公開する方法
- A2A クライアントを使用して A2A サーバーをテストする
必要なもの
2. 始める前に
プロジェクトを作成する
- Google Cloud コンソールのプロジェクト選択ページで、Google Cloud プロジェクトを選択または作成します。
- Cloud プロジェクトに対して課金が有効になっていることを確認します。プロジェクトで課金が有効になっているかどうかを確認する方法をご覧ください。
- このリンクをクリックして Cloud Shell を有効にします。Cloud Shell の対応するボタンをクリックして、Cloud Shell ターミナル（クラウド コマンドの実行用）とエディタ（プロジェクトのビルド用）を切り替えることができます。
- Cloud Shell に接続したら、次のコマンドを使用して、認証が完了していることと、プロジェクトがプロジェクト ID に設定されていることを確認します。
gcloud auth list
- Cloud Shell で次のコマンドを実行して、gcloud コマンドがプロジェクトを認識していることを確認します。
gcloud config list project
- 次のコマンドを使用してプロジェクトを設定します。
export PROJECT_ID=<YOUR_PROJECT_ID>
gcloud config set project $PROJECT_ID
- 次のコマンドを使用して、必要な API を有効にします。これには数分かかることがあります。
gcloud services enable cloudresourcemanager.googleapis.com \
servicenetworking.googleapis.com \
run.googleapis.com \
cloudbuild.googleapis.com \
artifactregistry.googleapis.com \
aiplatform.googleapis.com \
compute.googleapis.com
- Python 3.10 以降がインストールされていることを確認します。
gcloud コマンドとその使用方法については、ドキュメントをご覧ください。
3. インストール
- リポジトリのクローンを作成します。
git clone https://github.com/jackwotherspoon/currency-agent.git
cd currency-agent
- uv（依存関係の管理に使用）をインストールします。
# macOS and Linux
curl -LsSf https://astral.sh/uv/install.sh | sh
# Windows (uncomment below line)
# powershell -ExecutionPolicy ByPass -c "irm https://astral.sh/uv/install.ps1 | iex"
注 「uv が見つかりません」というエラーが表示された場合は、uv をインストールした後にターミナルを再起動するか、新しいターミナルを開く必要があります。
- 環境変数を構成する（
.envファイル経由）:
次のコマンドを実行して .env ファイルを作成します。
echo "GOOGLE_GENAI_USE_VERTEXAI=TRUE" >> .env \
&& echo "GOOGLE_CLOUD_PROJECT=$PROJECT_ID" >> .env \
&& echo "GOOGLE_CLOUD_LOCATION=us-central1" >> .env
4. ローカル MCP サーバーを作成する
通貨エージェントのオーケストレーションを行う前に、エージェントが必要とするツールを公開する MCP サーバーを作成します。
MCP サーバーを使用すると、軽量なプログラムを記述して、特定の機能（為替レートの取得など）をツールとして公開できます。エージェントまたは複数のエージェントは、標準化された Model Context Protocol（MCP）を使用してこれらのツールにアクセスできます。
FastMCP Python パッケージを使用して、get_exchange_rate という単一のツールを公開する MCP サーバーを作成できます。get_exchange_rate ツールは、インターネット経由で Frankfurter API を呼び出し、2 つの通貨間の現在の為替レートを取得します。
MCP サーバーのコードは mcp-server/server.py ファイルにあります。
import logging
import os
import httpx
from fastmcp import FastMCP
# Set up logging
logger = logging.getLogger(__name__)
logging.basicConfig(format="[%(levelname)s]: %(message)s", level=logging.INFO)
mcp = FastMCP("Currency MCP Server 💵")
@mcp.tool()
def get_exchange_rate(
currency_from: str = 'USD',
currency_to: str = 'EUR',
currency_date: str = 'latest',
):
"""Use this to get current exchange rate.
Args:
currency_from: The currency to convert from (e.g., "USD").
currency_to: The currency to convert to (e.g., "EUR").
currency_date: The date for the exchange rate or "latest". Defaults to "latest".
Returns:
A dictionary containing the exchange rate data, or an error message if the request fails.
"""
logger.info(f"--- 🛠️ Tool: get_exchange_rate called for converting {currency_from} to {currency_to} ---")
try:
response = httpx.get(
f'https://api.frankfurter.app/{currency_date}',
params={'from': currency_from, 'to': currency_to},
)
response.raise_for_status()
data = response.json()
if 'rates' not in data:
return {'error': 'Invalid API response format.'}
logger.info(f'✅ API response: {data}')
return data
except httpx.HTTPError as e:
return {'error': f'API request failed: {e}'}
except ValueError:
return {'error': 'Invalid JSON response from API.'}
if __name__ == "__main__":
logger.info(f"🚀 MCP server started on port {os.getenv('PORT', 8080)}")
# Could also use 'sse' transport, host="0.0.0.0" required for Cloud Run.
asyncio.run(
mcp.run_async(
transport="streamable-http",
host="0.0.0.0",
port=os.getenv("PORT", 8080),
)
)
MCP サーバーをローカルで起動するには、ターミナルを開いて次のコマンドを実行します（サーバーは http://localhost:8080 で起動します）。
uv run mcp-server/server.py
MCP サーバーが正常に機能し、Model Context Protocol を使用して get_exchange_rate ツールにアクセスできることをテストします。
ローカル MCP サーバーを停止しないように、新しいターミナル ウィンドウで次のコマンドを実行します。
uv run mcp-server/test_server.py
1 USD（米ドル）と EUR（ユーロ）の現在の為替レートが出力されます。
--- 🛠️ Tool found: get_exchange_rate ---
--- 🪛 Calling get_exchange_rate tool for USD to EUR ---
--- ✅ Success: {
"amount": 1.0,
"base": "USD",
"date": "2025-05-26",
"rates": {
"EUR": 0.87866
}
} ---
、エージェントがアクセスできるツールを備えた、正常に動作する MCP サーバーが作成されました。
次のステップに進む前に、起動したターミナルで Ctrl+C（Mac の場合は Command+C）を実行して、ローカルで実行されている MCP サーバーを停止します。
5. MCP サーバーを Cloud Run にデプロイする
これで、MCP サーバーをリモート MCP サーバーとして Cloud Run にデプロイする準備が整いました。🚀☁️
MCP サーバーをリモートで実行するメリット
Cloud Run で MCP サーバーをリモートで実行すると、次のようなメリットがあります。
- 📈スケーラビリティ: Cloud Run は、受信したすべてのリクエストを迅速に処理するためにスケールアウトするように構築されています。Cloud Run は、需要に応じて MCP サーバーを自動的にスケーリングします。
- 👥一元化されたサーバー: IAM 権限を使用して、一元化された MCP サーバーのアクセス権をチームメンバーと共有できます。これにより、チームメンバーは、独自のサーバーをローカルで実行するのではなく、ローカルマシンからサーバーに接続できます。MCP サーバーに変更を加えると、すべてのチームメンバーがその変更の恩恵を受けることができます。
- 🔐セキュリティ: Cloud Run では、認証済みリクエストを強制的に行う簡単な方法が用意されています。これにより、MCP サーバーに安全な接続のみが許可され、不正アクセスを防ぐことができます。
重要: 前述のセキュリティに関する点は重要です。認証を適用しないと、一般公開されているインターネット上のすべてのユーザーが MCP サーバーにアクセスして呼び出す可能性があります。
mcp-server ディレクトリに移動します。
cd mcp-server
MCP サーバーを Cloud Run にデプロイします。
gcloud run deploy mcp-server --no-allow-unauthenticated --region=us-central1 --source .
サービスが正常にデプロイされると、次のようなメッセージが表示されます。
Service [mcp-server] revision [mcp-server-12345-abc] has been deployed and is serving 100 percent of traffic.
MCP クライアントの認証
--no-allow-unauthenticated を指定して認証を必須にしたので、リモート MCP サーバーに接続する MCP クライアントは認証する必要があります。
MCP クライアントを実行している場所に応じて、Cloud Run で MCP サーバーをホストするの公式ドキュメントでこのトピックの詳細を確認してください。
Cloud Run プロキシを実行して、ローカルマシンでリモート MCP サーバーへの認証済みトンネルを作成する必要があります。
デフォルトでは、Cloud Run サービスの URL では、すべてのリクエストに Cloud Run 起動元（roles/run.invoker）IAM ロールによる承認が必要です。この IAM ポリシー バインディングにより、強力なセキュリティ メカニズムを使用してローカル MCP クライアントを認証できます。
リモート MCP サーバーにアクセスしようとしているユーザーまたはチーム メンバーに、IAM プリンシパル（Google Cloud アカウント）にバインドされた roles/run.invoker IAM ロールがあることを確認する必要があります。
注: 次のコマンドを実行すると、Cloud Run プロキシがまだインストールされていない場合は、プロキシのダウンロードを求めるメッセージが表示されることがあります。画面の指示に沿ってダウンロードしてインストールします。
gcloud run services proxy mcp-server --region=us-central1
次の出力が表示されます。
Proxying to Cloud Run service [mcp-server] in project [<YOUR_PROJECT_ID>] region [us-central1]
http://127.0.0.1:8080 proxies to https://mcp-server-abcdefgh-uc.a.run.app
これで、http://127.0.0.1:8080 へのすべてのトラフィックが認証され、リモート MCP サーバーに転送されるようになります。
リモート MCP サーバーをテストする
新しいターミナルでルート フォルダに戻り、mcp-server/test_server.py ファイルを再度実行して、リモート MCP サーバーが動作していることを確認します。
注: Codelab の残りの部分では、Cloud Run プロキシを実行したままにしてください。
cd ..
uv run mcp-server/test_server.py
ローカルでサーバーを実行した場合と同様の出力が表示されます。
--- 🛠️ Tool found: get_exchange_rate ---
--- 🪛 Calling get_exchange_rate tool for USD to EUR ---
--- ✅ Success: {
"amount": 1.0,
"base": "USD",
"date": "2025-05-26",
"rates": {
"EUR": 0.87866
}
} ---
リモート サーバーが実際に呼び出されたことを確認するには、デプロイされた Cloud Run MCP サーバーのログをクエリします。
gcloud run services logs read mcp-server --region us-central1 --limit 5
ログに次のように出力されます。
2025-06-04 14:28:29,871 [INFO]: --- 🛠️ Tool: get_exchange_rate called for converting USD to EUR ---
2025-06-04 14:28:30,610 [INFO]: HTTP Request: GET https://api.frankfurter.app/latest?from=USD&to=EUR "HTTP/1.1 200 OK"
2025-06-04 14:28:30,611 [INFO]: ✅ API response: {'amount': 1.0, 'base': 'USD', 'date': '2025-06-03', 'rates': {'EUR': 0.87827}}
リモート MCP サーバーが作成されたので、エージェントの作成に進みます。🤖
6. エージェント開発キット（ADK）を使用してエージェントを作成する
MCP サーバーがデプロイされたので、エージェント開発キット（ADK）を使用して通貨エージェントを作成します。
Agent Development Kit の v1.0.0 安定版が最近リリースされました。このマイルストーンは、Python ADK が本番環境に対応し、デベロッパーが本番環境でエージェントを自信を持って構築してデプロイできる、信頼性が高く堅牢なプラットフォームを提供できることを意味します。
ADK を使用すると、エージェントの作成が非常に軽量になり、MCP ツールのサポートが組み込まれているため、MCP サーバーに簡単に接続できます。通貨エージェントは、ADK の MCPToolset クラスを使用して get_exchange_rate ツールにアクセスします。
通貨エージェントのコードは currency_agent/agent.py にあります。
import logging
import os
from dotenv import load_dotenv
from google.adk.agents import LlmAgent
from google.adk.tools.mcp_tool.mcp_toolset import MCPToolset, SseServerParams
logger = logging.getLogger(__name__)
logging.basicConfig(format="[%(levelname)s]: %(message)s", level=logging.INFO)
load_dotenv()
SYSTEM_INSTRUCTION = (
'You are a specialized assistant for currency conversions. '
"Your sole purpose is to use the 'get_exchange_rate' tool to answer questions about currency exchange rates. "
'If the user asks about anything other than currency conversion or exchange rates, '
'politely state that you cannot help with that topic and can only assist with currency-related queries. '
'Do not attempt to answer unrelated questions or use tools for other purposes.'
)
def create_agent() -> LlmAgent:
"""Constructs the ADK currency conversion agent."""
logger.info("--- 🔧 Loading MCP tools from MCP Server... ---")
logger.info("--- 🤖 Creating ADK Currency Agent... ---")
return LlmAgent(
model='gemini-2.0-flash-001',
name='currency_agent',
description="An agent that can help with currency conversions",
instruction=SYSTEM_INSTRUCTION,
tools=[
MCPToolset(
connection_params=SseServerParams(
url=os.getenv('MCP_SERVER_URL', 'http://localhost:8080/sse')
)
)
],
)
root_agent = create_agent()
通貨エージェントをすばやくテストするには、ADK のデベロッパー UI を使用します。この UI には、adk web を実行してアクセスできます。
uv run adk web
ブラウザで http://localhost:8000 にアクセスして、エージェントを表示してテストしましょう。
ウェブ UI の左上にあるエージェントとして currency_agent が選択されていることを確認します。
チャット欄でエージェントに「250 カナダドルを米ドルに換算するといくらですか？」などのように質問します。エージェントが get_exchange_rate MCP ツールを呼び出してから、回答を返すはずです。
エージェントは機能します。通貨換算に関するクエリを処理できます。
7. Agent2Agent（A2A）プロトコル
Agent2Agent（A2A）プロトコルは、AI エージェント間のシームレスな通信とコラボレーションを可能にするために設計されたオープン スタンダードです。これにより、さまざまなフレームワークを使用して異なるベンダーが構築したエージェントが、共通の言語で相互に通信できるため、サイロを解消し、相互運用性を促進できます。
A2A を使用すると、エージェントは次のことができます。
- 検索: 標準化されたエージェントカードを使用して、他のエージェントを見つけ、そのスキル（AgentSkill）と機能（AgentCapabilities）を確認できます。
- 通信: メッセージとデータを安全に交換します。
- コラボレーション: タスクを委任し、アクションを調整して複雑な目標を達成します。
A2A プロトコルは、「エージェントカード」などのメカニズムを通じてこの通信を容易にします。エージェントカードは、エージェントが自分の機能や接続情報を宣伝するために使用できるデジタル名刺のようなものです。
次に、A2A を使用して通貨エージェントを公開し、他のエージェントやクライアントから呼び出せるようにします。
A2A Python SDK
A2A Python SDK には、前述の各リソース（AgentSkill、AgentCapabilities、AgentCard）の Pydantic モデルが用意されています。これにより、A2A プロトコルとの開発と統合を迅速に行うためのインターフェースが提供されます。
AgentSkill は、通貨エージェントが get_exchange_rate 用のツールを持っていることを他のエージェントに宣伝する方法です。
# A2A Agent Skill definition
skill = AgentSkill(
id='get_exchange_rate',
name='Currency Exchange Rates Tool',
description='Helps with exchange values between various currencies',
tags=['currency conversion', 'currency exchange'],
examples=['What is exchange rate between USD and GBP?'],
)
次に、AgentCard の一部として、エージェントのスキルと機能が、エージェントが処理できる入出力モードなどの追加情報とともに一覧表示されます。
# A2A Agent Card definition
agent_card = AgentCard(
name='Currency Agent',
description='Helps with exchange rates for currencies',
url=f'http://{host}:{port}/',
version='1.0.0',
defaultInputModes=["text"],
defaultOutputModes=["text"],
capabilities=AgentCapabilities(streaming=True),
skills=[skill],
)
AgentExecutor インターフェースは、A2A エージェントがリクエストを処理してレスポンスやイベントを生成するコアロジックを処理します。A2A Python SDK には、実装が必要な抽象基底クラス a2a.server.agent_execution.AgentExecutor が用意されています。
通貨エージェントと連携して A2A の威力を発揮しましょう。
8. 通貨エージェント A2A サーバー
次に、いくつかのコードを見て、A2A サーバーを構成するさまざまな部分がどのように組み合わされているかを確認します。
ファイル currency_agent/agent_executor.py には、A2A 抽象クラス AgentExecutor から継承するクラス ADKAgentExecutor があります。ADK ランナーを呼び出し、エージェントへのリクエストを処理し、ADK が使用する google.genai.types と A2A が使用する a2a.types を相互に変換することで、ADK エージェントの呼び出しを処理します。
# ... see file for full code
class ADKAgentExecutor(AgentExecutor):
"""An AgentExecutor that runs an ADK agent."""
def __init__(self, runner: Runner, card: AgentCard):
self.runner = runner
self._card = card
self._running_sessions = {}
def _run_agent(
self, session_id, new_message: types.Content
) -> AsyncGenerator[Event, None]:
return self.runner.run_async(
session_id=session_id, user_id="self", new_message=new_message
)
async def _process_request(
self,
new_message: types.Content,
session_id: str,
task_updater: TaskUpdater,
) -> None:
session = await self._upsert_session(
session_id,
)
session_id = session.id
# Run through all events within the request.
async for event in self._run_agent(session_id, new_message):
if event.is_final_response():
parts = convert_genai_parts_to_a2a(event.content.parts)
logger.debug('✅ Yielding final response: %s', parts)
task_updater.add_artifact(parts)
task_updater.complete()
break
# If the agent is not making a function call, yield an update.
if not event.get_function_calls():
logger.debug('⏳ Yielding update response')
task_updater.update_status(
TaskState.working,
message=task_updater.new_agent_message(
convert_genai_parts_to_a2a(event.content.parts),
),
)
else:
logger.debug('➡️ Skipping event')
async def execute(
self,
context: RequestContext,
event_queue: EventQueue,
):
# Run the agent until either complete or the task is suspended.
updater = TaskUpdater(event_queue, context.task_id, context.context_id)
# Immediately notify that the task is submitted.
if not context.current_task:
updater.submit()
updater.start_work()
await self._process_request(
types.UserContent(
parts=convert_a2a_parts_to_genai(context.message.parts),
),
context.context_id,
updater,
)
logger.debug('--- 💵💱💶 [Currency] execute exiting ---')
# ... see file for full code
currency_agent/__main__.py 内で、AgentSkill と AgentCard を初期化し、ADK 通貨エージェントを作成します。また、A2A サーバーの設定と起動も行います。
A2A Python SDK には、A2A 準拠の HTTP サーバーの実行を簡素化する A2AStarletteApplication クラスが用意されています。ウェブ フレームワークに Starlette を使用し、通常は Uvicorn などの ASGI サーバーで実行されます。
# ... see file for full code
@click.command()
@click.option('--host', 'host', default='localhost')
@click.option('--port', 'port', default=10000)
def main(host: str, port: int):
# Verify one of Google AI Studio or Vertex AI is being used
if os.getenv("GOOGLE_GENAI_USE_VERTEXAI") != "TRUE" and not os.getenv(
"GOOGLE_API_KEY"
):
raise ValueError(
"GOOGLE_API_KEY environment variable not set and "
"GOOGLE_GENAI_USE_VERTEXAI is not TRUE."
)
# A2A Agent Skill definition
skill = AgentSkill(
id='get_exchange_rate',
name='Currency Exchange Rates Tool',
description='Helps with exchange values between various currencies',
tags=['currency conversion', 'currency exchange'],
examples=['What is exchange rate between USD and GBP?'],
)
# A2A Agent Card definition
agent_card = AgentCard(
name='Currency Agent',
description='Helps with exchange rates for currencies',
url=f'http://{host}:{port}/',
version='1.0.0',
defaultInputModes=["text"],
defaultOutputModes=["text"],
capabilities=AgentCapabilities(streaming=True),
skills=[skill],
)
# Create the ADK runner and executor.
runner = Runner(
app_name=agent_card.name,
agent=root_agent,
artifact_service=InMemoryArtifactService(),
session_service=InMemorySessionService(),
memory_service=InMemoryMemoryService(),
)
agent_executor = ADKAgentExecutor(runner, agent_card)
request_handler = DefaultRequestHandler(
agent_executor=agent_executor,
task_store=InMemoryTaskStore(),
)
server = A2AStarletteApplication(
agent_card=agent_card, http_handler=request_handler
)
uvicorn.run(server.build(), host=host, port=port)
# ... see file for full code
A2A サーバーを実行するには、新しいターミナルで次のコマンドを実行します。
uv run currency_agent/
サーバーが正常に起動すると、ポート 10000 で実行されていることを示す次のような出力が表示されます。
[INFO]: --- 🔧 Loading MCP tools from MCP Server... ---
[INFO]: --- 🤖 Creating ADK Currency Agent... ---
INFO: Started server process [45824]
INFO: Waiting for application startup.
INFO: Application startup complete.
INFO: Uvicorn running on http://localhost:10000 (Press CTRL+C to quit)
通貨エージェントが A2A サーバーとして正常に実行され、A2A プロトコルを使用して他のエージェントまたはクライアントから呼び出せるようになりました。
A2A サーバーをテストする
A2A を使用してリクエストを送信して、サーバーをテストできます。
A2A Python SDK には、この作業を簡素化する a2a.client.A2AClient クラスが用意されています。
注: マルチエージェント アーキテクチャを使用している場合は、ここで A2AClient で行ったものと同様のパターンを使用します。
ファイル currency_agent/test_client.py には、A2A サーバーを対象としたさまざまなテストケースを実行するコードが含まれています。
# ... see file for full code
# Example test using A2AClient
async def run_single_turn_test(client: A2AClient) -> None:
"""Runs a single-turn non-streaming test."""
send_payload = create_send_message_payload(
text='how much is 100 USD in CAD?'
)
request = SendMessageRequest(params=MessageSendParams(**send_payload))
print('--- ✉️ Single Turn Request ---')
# Send Message
send_response: SendMessageResponse = await client.send_message(request)
print_json_response(send_response, '📥 Single Turn Request Response')
if not isinstance(send_response.root, SendMessageSuccessResponse):
print('received non-success response. Aborting get task ')
return
if not isinstance(send_response.root.result, Task):
print('received non-task response. Aborting get task ')
return
task_id: str = send_response.root.result.id
print('--- ❔ Query Task ---')
# query the task
get_request = GetTaskRequest(params=TaskQueryParams(id=task_id))
get_response: GetTaskResponse = await client.get_task(get_request)
print_json_response(get_response, '📥 Query Task Response')
# ----- Main Entrypoint (Create client --> Run tests) -----
async def main() -> None:
"""Main function to run the tests."""
print(f'--- 🔄 Connecting to agent at {AGENT_URL}... ---')
try:
async with httpx.AsyncClient() as httpx_client:
client = await A2AClient.get_client_from_agent_card_url(
httpx_client, AGENT_URL
)
print('--- ✅ Connection successful. ---')
await run_single_turn_test(client)
await run_streaming_test(client)
await run_multi_turn_test(client)
except Exception as e:
traceback.print_exc()
print(f'--- ❌ An error occurred: {e} ---')
print('Ensure the agent server is running.')
次のコマンドを使用してテストを実行します。
uv run currency_agent/test_client.py
テスト実行が成功すると、次の結果が得られます。
--- 🔄 Connecting to agent at http://localhost:10000... ---
--- ✅ Connection successful. ---
--- ✉️ Single Turn Request ---
--- 📥 Single Turn Request Response ---
{"id":"3bc92d7b-d857-4e93-9ff0-b2fb865f6e35","jsonrpc":"2.0","result":{"artifacts":[{"artifactId":"35e89e14-b977-4397-a23b-92c84bc32379","parts":[{"kind":"text","text":"Based on the current exchange rate, 1 USD is equivalent to 1.3704 CAD. Therefore, 100 USD would be 137.04 CAD.\n"}]}],"contextId":"2d66f277-152c-46ef-881d-7fe32866e9f5","history":[{"contextId":"2d66f277-152c-46ef-881d-7fe32866e9f5","kind":"message","messageId":"59819269f7d04849b0bfca7d43ec073c","parts":[{"kind":"text","text":"how much is 100 USD in CAD?"}],"role":"user","taskId":"52ae2392-84f5-429a-a14b-8413d3d20d97"},{"contextId":"2d66f277-152c-46ef-881d-7fe32866e9f5","kind":"message","messageId":"286095c6-12c9-40cb-9596-a9676d570dbd","parts":[],"role":"agent","taskId":"52ae2392-84f5-429a-a14b-8413d3d20d97"}],"id":"52ae2392-84f5-429a-a14b-8413d3d20d97","kind":"task","status":{"state":"completed"}}}
// ...
--- ⏩ Single Turn Streaming Request ---
--- ⏳ Streaming Chunk ---
{"id":"21239a5f-abbf-4a5e-a249-c101eb1dfbdd","jsonrpc":"2.0","result":{"contextId":"f268ad8c-b3bf-4439-9a64-5e02dfbb9a62","final":false,"kind":"status-update","status":{"state":"submitted"},"taskId":"761d2275-d58b-46f8-9c8d-68cd72e0667d"}}
--- ⏳ Streaming Chunk ---
{"id":"21239a5f-abbf-4a5e-a249-c101eb1dfbdd","jsonrpc":"2.0","result":{"contextId":"f268ad8c-b3bf-4439-9a64-5e02dfbb9a62","final":false,"kind":"status-update","status":{"state":"working"},"taskId":"761d2275-d58b-46f8-9c8d-68cd72e0667d"}}
--- ⏳ Streaming Chunk ---
{"id":"21239a5f-abbf-4a5e-a249-c101eb1dfbdd","jsonrpc":"2.0","result":{"contextId":"f268ad8c-b3bf-4439-9a64-5e02dfbb9a62","final":false,"kind":"status-update","status":{"message":{"contextId":"f268ad8c-b3bf-4439-9a64-5e02dfbb9a62","kind":"message","messageId":"25f5f972-9475-4e4a-a08d-e13f521d7462","parts":[],"role":"agent","taskId":"761d2275-d58b-46f8-9c8d-68cd72e0667d"},"state":"working"},"taskId":"761d2275-d58b-46f8-9c8d-68cd72e0667d"}}
--- ⏳ Streaming Chunk ---
{"id":"21239a5f-abbf-4a5e-a249-c101eb1dfbdd","jsonrpc":"2.0","result":{"artifact":{"artifactId":"35e89e14-b977-4397-a23b-92c84bc32379","parts":[{"kind":"text","text":"The current exchange rate is 1 EUR to 164.15 JPY. So, 50 EUR would be 8207.5 JPY.\n"}]},"contextId":"f268ad8c-b3bf-4439-9a64-5e02dfbb9a62","kind":"artifact-update","taskId":"761d2275-d58b-46f8-9c8d-68cd72e0667d"}}
// ...
--- 🚀 First turn completed, no further input required for this test case. ---
うまくいったA2A サーバー経由で通貨エージェントと通信できることが確認できました。🎉
より高度なユースケースについては、GitHub の a2a-samples リポジトリをご覧ください。
エージェントをデプロイする方法については、Vertex AI Agent Engine は、AI エージェントを本番環境にデプロイするためのマネージド エクスペリエンスを提供します。
9. 完了
これで、これで、リモート MCP サーバーの構築とデプロイが完了し、MCP を使用してツールに接続するエージェント開発キット（ADK）を使用して通貨エージェントを作成し、Agent2Agent（A2A）プロトコルを使用してエージェントを公開できました。通貨エージェントが A2A を使用して、任意のフレームワークの他のエージェントとやり取りできるようになりました。
コード全体のドキュメントへのリンクはこちらです。
学習した内容
- ローカル MCP サーバーを作成する方法
- MCP サーバーを Cloud Run にデプロイする
- MCP ツールを使用するエージェント開発キットを使用してエージェントを構築する方法
- ADK エージェントを A2A サーバーとして公開する方法
- A2A クライアントを使用して A2A サーバーをテストする
クリーンアップ
このラボで使用したリソースについて、Google Cloud アカウントに課金されないようにするには、次の操作を行います。
- Google Cloud コンソールで、[リソースの管理] ページに移動します。
- プロジェクト リストで、削除するプロジェクトを選択し、[削除] をクリックします。
- ダイアログでプロジェクト ID を入力し、[シャットダウン] をクリックしてプロジェクトを削除します。