1. はじめに
AI エージェントがより多くの責任を負うようになると、1 つのエージェントがすべてを行うことは、維持、スケーリング、進化が困難になります。機能が異なると、デプロイ戦略、更新サイクル、所有するチームが異なることがよくあります。
- A2A（Agent2Agent）プロトコルは、通信側の問題を解決します。エージェントが互いの機能を検出し、フレームワークや組織を越えて連携する方法を標準化します。
- Gemini Enterprise Agent Platform Runtime は、デプロイ側の問題を解決します。これは、A2A サポート、自動スケーリング、安全なエンドポイント、永続セッション、インフラストラクチャ管理の不要なエージェントをホストするフルマネージドのサーバーレス プラットフォームです。
これらを組み合わせることで、特化型エージェントを構築し、検出可能な A2A サービスとしてデプロイして、マルチエージェント システムに構成できます。
作成するアプリの概要
Gemini Enterprise Agent Platform Sessions によって管理される ADK セッション状態を使用して、レストランのテーブル予約（作成、確認、キャンセル）を管理する予約エージェント。このエージェントを Gemini Enterprise Agent Platform ランタイムにデプロイすると、A2A プロトコルのエージェント カードを介して検出可能になります。次に、Foodie Finds レストラン コンシェルジュ エージェント（前提条件の Codelab から。Codelab をまだご覧になっていない場合でも、スターター リポジトリをご用意していますのでご安心ください）をアップグレードして、予約エージェントをリモート A2A サブエージェントとして使用します。その結果、オーケストレーターがメニュークエリを MCP ツールボックスに、予約リクエストをリモート A2A エージェントにルーティングするマルチエージェント システムが作成されます。
学習内容
- マネージド セッション サービスを使用して予約データを管理する ADK エージェントを構築する
- エージェント カードとスキルを使用して ADK エージェントを A2A サーバーとして公開する
- Gemini Enterprise エージェント ランタイムに A2A エージェントをデプロイする
RemoteA2aAgentを使用して別の ADK エージェントからリモート A2A エージェントを使用し、認証されたリクエストを処理する- マルチエージェント システムを段階的にテストする: ローカル A2A、デプロイされた A2A、部分統合、完全なデプロイ
前提条件
- （推奨）次の Codelab を完了していること :
- ADK と CloudSQL を使用して永続的な AI エージェントを構築する -> ADK セッションと状態の詳細
- ADK、MCP ツールボックス、Cloud SQL を使用したエージェント RAG -> この Codelab でエージェントの構築を続行できます。提供されるスターター コードは同じです。
- 有効な請求先アカウントを持つ Google Cloud アカウント
- Python と ADK のコンセプトに関する基本的な知識
2. 環境のセットアップ - 前の Codelab からの継続
この Codelab で提供する説明は、前提条件の Codelab: ADK、MCP ツールボックス、Cloud SQL を使用したエージェント RAG の続きです。前の Codelab から作業を続行できます
前の Codelab の作業ディレクトリ（作業ディレクトリは build-agent-adk-toolbox-cloudsql になります）でビルドを開始できます。混乱を避けるため、ディレクトリの名前を、最初からやり直すときに使用するディレクトリ名に変更しましょう。
mv ~/build-agent-adk-toolbox-cloudsql ~/adk-a2a-agent-runtime-starter
cloudshell workspace ~/adk-a2a-agent-runtime-starter && cd ~/adk-a2a-agent-runtime-starter
source .env
前の Codelab の鍵ファイルが配置されていることを確認します。
echo "--- Restaurant Agent ---"
cat restaurant_agent/agent.py | head -5
echo ""
echo "--- Toolbox Config ---"
cat tools.yaml | head -5
LlmAgent インポートを含む restaurant_agent/agent.py ファイルと、Toolbox 構成を含む tools.yaml が表示されます。
次に、Python 環境を再初期化しましょう。
rm -rf .venv
uv sync
また、データベースがシードされ、準備ができていることを確認します。
uv run python scripts/verify_seed.py
前の Codelab のテストの詳細に沿ってテストを行うと、次のような出力が表示されることがあります。
Menu Items: 16/15 Embeddings: 16/15 ✗ Database not ready
大丈夫です。データベース チェックでは、データの取り込みチェックから入力した追加データは考慮されません。データが 15 個以上あれば問題ありません。
必要な API を有効にする
次に、Gemini Enterprise Agent Platform とやり取りするために必要な API を有効にする必要があります。
gcloud services enable \
cloudresourcemanager.googleapis.com
次のセクションに進むために必要なファイルとインフラストラクチャはすでに用意されているはずです。A2A Protocol and Gemini Enterprise Agent Runtime
3. 環境のセットアップ - スターター リポジトリで新たに開始する
このステップでは、Cloud Shell 環境を準備し、Google Cloud プロジェクトを構成して、スターター リポジトリのクローンを作成します。
Cloud Shell を開く
ブラウザで Cloud Shell を開きます。Cloud Shell には、この Codelab で必要なすべてのツールがプリインストールされた環境が用意されています。プロンプトが表示されたら、[Authorize] をクリックします。
[表示] -> [ターミナル] をクリックしてターミナルを開きます。インターフェースは次のようになります。
これがメイン インターフェースになります。上部に IDE、下部にターミナルが表示されます。
作業ディレクトリを設定する
スターター リポジトリのクローンを作成します。この Codelab で作成するすべてのコードはここに保存されます。
rm -rf ~/adk-a2a-agent-runtime-starter
git clone https://github.com/alphinside/adk-a2a-agent-runtime-starter.git
cloudshell workspace ~/adk-a2a-agent-runtime-starter && cd ~/adk-a2a-agent-runtime-starter
提供されたテンプレートから .env ファイルを作成します。
cp .env.example .env
ターミナルでのプロジェクト設定を簡素化するには、このプロジェクト設定スクリプトを作業ディレクトリにダウンロードします。
curl -sL https://raw.githubusercontent.com/alphinside/cloud-trial-project-setup/main/setup_verify_trial_project.sh -o setup_verify_trial_project.sh
スクリプトを実行します。トライアルの請求先アカウントを確認し、新しいプロジェクトを作成（または既存のプロジェクトを検証）し、プロジェクト ID を現在のディレクトリの .env ファイルに保存し、gcloud でアクティブ プロジェクトを設定します。
bash setup_verify_trial_project.sh && source .env
このスクリプトによって行われる処理は次のとおりです。
- 有効なトライアルの請求先アカウントがあることを確認する
.envに既存のプロジェクトがあるかどうかを確認します（ある場合）。- 新しいプロジェクトを作成するか、既存のプロジェクトを再利用する
- トライアルの請求先アカウントをプロジェクトにリンクする
- プロジェクト ID を
.envに保存する - プロジェクトをアクティブな
gcloudプロジェクトとして設定する
Cloud Shell ターミナルのプロンプトで、作業ディレクトリの横にある黄色のテキストを確認して、プロジェクトが正しく設定されていることを確認します。プロジェクト ID が表示されます。
必要な API を有効にする
次に、Gemini Enterprise Agent Platform とやり取りするために必要な API を有効にする必要があります。
gcloud services enable \
aiplatform.googleapis.com \
cloudresourcemanager.googleapis.com
スターター インフラストラクチャの設定
まず、uv を使用して Python の依存関係をインストールする必要があります。これは、Rust で記述された高速な Python パッケージとプロジェクト マネージャーです（uv ドキュメント）。この Codelab では、Python プロジェクトの保守を高速かつ簡単に行うために使用します。
uv sync
次に、フル設定スクリプトを実行します。このスクリプトは、Cloud SQL インスタンスを作成し、データをシードして、レストラン エージェントの初期状態として機能する Toolbox サービスをデプロイします。
bash scripts/full_setup.sh > logs/full_setup.log 2>&1 &
4. コンセプト: Agent2Agent（A2A）プロトコルと Gemini Enterprise エージェント ランタイム
構築する前に、この Codelab でエージェント アプリケーションをスケーリングするために使用する 2 つの主要なテクノロジーについて簡単に説明します。
Agent2Agent（A2A）プロトコル
Agent2Agent（A2A）プロトコルは、AI エージェント間のシームレスな通信とコラボレーションを可能にするように設計されたオープン スタンダードです。MCP（Model Context Protocol） がエージェントをツールやデータに接続するのに対し、A2A はエージェントを他のエージェントに接続します。これにより、エージェントは互いの機能を検出し、タスクを委任し、フレームワークや組織を越えて連携できます。
エージェントをツールとして（MCP 経由で）ラップする場合と、A2A 経由で公開する場合の主な違いは、ツールはステートレスで単一の機能を実行するのに対し、A2A エージェントは推論、状態の維持、交渉や説明などの複数ターンのやり取りを処理できることです。A2A 経由で公開されたエージェントは、関数呼び出しに縮小されるのではなく、完全な機能を保持します。
A2A では、次の 3 つの基本コンセプトが定義されています。
- エージェント カード - エージェントの役割、スキル、エンドポイントを記述する JSON ドキュメント。他のエージェントはこのカードを取得して、機能を確認します。
- メッセージ - A2A エンドポイントに送信され、タスクをトリガーするユーザーまたはエージェントのリクエスト。
- タスク - ライフサイクル（送信 → 処理中 → 完了/失敗）と結果を含むアーティファクトを持つ作業単位。
詳細については、A2A とはをご覧ください。
Gemini Enterprise Agent Platform Runtime
Agent Runtime は、Google Cloud 上のフルマネージド サービスで、エンタープライズ セキュリティ機能（VPC Service Controls、CMEK など）を使用して本番環境で AI エージェントをデプロイ、スケーリング、管理できます。インフラストラクチャを処理してくれるため、ユーザーはエージェント ロジックに集中できます。
Agent Runtime には次の機能があります。
- マネージド デプロイ - ADK、LangGraph、または任意の Python フレームワークで構築されたエージェントを 1 回の SDK 呼び出しでデプロイします。
- A2A ホスティング - エージェントを A2A 準拠のエンドポイントとしてデプロイし、エージェント カードの自動サービングと認証済みアクセスを実現します。
- 永続セッション -
VertexAiSessionServiceは、リクエスト間で会話履歴と状態を保存します - 自動スケーリング - インフラストラクチャの管理なしで、トラフィックを処理するためにゼロからスケーリング
- オブザーバビリティ - Google Cloud のオブザーバビリティ スタックによる組み込みのトレース、ロギング、モニタリング
- その他の機能については、こちらのドキュメントをご覧ください。
この Codelab では、予約エージェントをエージェント ランタイムにデプロイします。デプロイプロセスでは、エージェント コードをシリアル化（ピクルス化）してアップロードします。Agent Runtime は、A2A プロトコルを処理するサーバーレス エンドポイントをプロビジョニングします。他のエージェント（またはクライアント）は、Google Cloud 認証情報で認証された標準の HTTP 呼び出しを介してこのエンドポイントとやり取りします。
5. 予約エージェントを構築する
このステップでは、セッション状態を使用してレストランの予約を処理する新しい ADK エージェントを作成します。エージェントは、電話番号をルックアップ キーとして、作成、確認、キャンセルという 3 つのオペレーションをサポートしています。すべての予約データは ADK のセッション状態に保存されます
エージェントをスキャフォールディングする
adk create を使用して、正しいモデルとプロジェクト構成でエージェント ディレクトリ構造を生成します。
source .env
uv run adk create reservation_agent \
--model gemini-2.5-flash \
--project ${GOOGLE_CLOUD_PROJECT} \
--region ${GOOGLE_CLOUD_LOCATION}
これにより、Agent Platform の Gemini モデル用に __init__.py、agent.py、.env が事前構成された reservation_agent/ ディレクトリが作成されます。
adk-a2a-agent-runtime-starter/ ├── reservation_agent/ │ ├── __init__.py │ ├── agent.py │ └── .env ├── logs ├── scripts └── ...
次に、エージェント コードを更新しましょう。
エージェント コードを記述する
生成されたエージェント ファイルを開きます。
cloudshell edit reservation_agent/agent.py
次に、内容を次のように置き換えます。
# reservation_agent/agent.py
from google.adk.agents import LlmAgent
from google.adk.tools import ToolContext
# App-scoped state prefix ensures reservations persist across all sessions.
# See https://adk.dev/sessions/state/ for state scope details.
STATE_PREFIX = "app:reservation:"
def create_reservation(
phone_number: str,
name: str,
party_size: int,
date: str,
time: str,
tool_context: ToolContext,
) -> dict:
"""Create a new restaurant reservation.
Args:
phone_number: Customer's phone number, used as the reservation ID.
name: Name for the reservation.
party_size: Number of guests.
date: Reservation date (e.g., '2025-07-15' or 'this Friday').
time: Reservation time (e.g., '7:00 PM').
Returns:
Confirmation of the reservation.
"""
reservation = {
"name": name,
"party_size": party_size,
"date": date,
"time": time,
"status": "confirmed",
}
tool_context.state[f"{STATE_PREFIX}{phone_number}"] = reservation
return {
"status": "confirmed",
"message": f"Reservation created for {name}, party of {party_size} on {date} at {time}. Phone: {phone_number}.",
}
def check_reservation(phone_number: str, tool_context: ToolContext) -> dict:
"""Look up an existing reservation by phone number.
Args:
phone_number: The phone number used when the reservation was created.
tool_context: ADK tool context for state access.
Returns:
The reservation details, or a message if not found.
"""
reservation = tool_context.state.get(f"{STATE_PREFIX}{phone_number}")
if reservation:
return {"found": True, "reservation": reservation}
return {"found": False, "message": f"No reservation found for {phone_number}."}
def cancel_reservation(phone_number: str, tool_context: ToolContext) -> dict:
"""Cancel an existing reservation by phone number.
Args:
phone_number: The phone number used when the reservation was created.
tool_context: ADK tool context for state access.
Returns:
Confirmation of cancellation, or a message if not found.
"""
key = f"{STATE_PREFIX}{phone_number}"
reservation = tool_context.state.get(key)
if not reservation:
return {"success": False, "message": f"No reservation found for {phone_number}."}
if reservation.get("status") == "cancelled":
return {"success": False, "message": f"Reservation for {phone_number} is already cancelled."}
reservation["status"] = "cancelled"
tool_context.state[key] = reservation
return {"success": True, "message": f"Reservation for {reservation['name']} ({phone_number}) has been cancelled."}
root_agent = LlmAgent(
name="reservation_agent",
model="gemini-2.5-flash",
instruction="""You are a friendly reservation assistant for "Foodie Finds" restaurant.
You help diners create, check, and cancel table reservations.
When a diner wants to make a reservation, collect these details:
- Name for the reservation
- Phone number (used as the reservation ID)
- Party size (number of guests)
- Date
- Time
Always confirm the details before creating the reservation.
When checking or cancelling, ask for the phone number if not provided.
Be concise and professional.""",
tools=[create_reservation, check_reservation, cancel_reservation],
)
6. A2A サーバー構成を準備する
A2A エージェントカードを定義する
エージェント カードは、エージェントの機能の構造化された説明です。他のエージェントとクライアントは、これを使用してエージェントの機能を検出します。カード構成を作成します。
cloudshell edit reservation_agent/a2a_config.py
以下を reservation_agent/a2a_config.py にコピーします。
# reservation_agent/a2a_config.py
from a2a.types import AgentSkill
from vertexai.preview.reasoning_engines.templates.a2a import create_agent_card
reservation_skill = AgentSkill(
id="manage_reservations",
name="Restaurant Reservations",
description="Create, check, and cancel table reservations at Foodie Finds restaurant",
tags=["reservations", "restaurant", "booking"],
examples=[
"Book a table for 4 on Friday at 7pm",
"Check reservation for 555-0101",
"Cancel my reservation, phone number 555-0101",
],
input_modes=["text/plain"],
output_modes=["text/plain"],
)
agent_card = create_agent_card(
agent_name="Reservation Agent",
description="Handles restaurant table reservations — create, check, and cancel bookings for Foodie Finds restaurant.",
skills=[reservation_skill],
)
A2A エグゼキュータを作成する
エグゼキュータは、A2A プロトコルと ADK エージェントをブリッジします。A2A リクエストを受信し、ADK エージェントを介して実行し、結果を A2A タスクとして返します。
cloudshell edit reservation_agent/executor.py
以下を reservation_agent/executor.py にコピーします。
# reservation_agent/executor.py
import os
from typing import NoReturn
import vertexai
from a2a.server.agent_execution import AgentExecutor, RequestContext
from a2a.server.events import EventQueue
from a2a.server.tasks import TaskUpdater
from a2a.types import TaskState, TextPart, UnsupportedOperationError
from a2a.utils import new_agent_text_message
from a2a.utils.errors import ServerError
from google.adk.artifacts import InMemoryArtifactService
from google.adk.memory.in_memory_memory_service import InMemoryMemoryService
from google.adk.runners import Runner
from google.adk.sessions import InMemorySessionService, VertexAiSessionService
from google.genai import types
from reservation_agent.agent import root_agent as reservation_agent
class ReservationAgentExecutor(AgentExecutor):
"""Bridge between the A2A protocol and the ADK reservation agent.
Uses InMemorySessionService for local testing, VertexAiSessionService
when deployed to Agent Runtime (detected via GOOGLE_CLOUD_AGENT_ENGINE_ID).
"""
def __init__(self) -> None:
self.agent = None
self.runner = None
def _init_agent(self) -> None:
if self.agent is not None:
return
self.agent = reservation_agent
engine_id = os.environ.get("GOOGLE_CLOUD_AGENT_ENGINE_ID")
if engine_id:
project = os.environ.get("GOOGLE_CLOUD_PROJECT")
location = os.environ.get("GOOGLE_CLOUD_LOCATION", "us-central1")
vertexai.init(project=project, location=location)
session_service = VertexAiSessionService(
project=project, location=location, agent_engine_id=engine_id,
)
app_name = engine_id
else:
session_service = InMemorySessionService()
app_name = self.agent.name
self.runner = Runner(
app_name=app_name,
agent=self.agent,
artifact_service=InMemoryArtifactService(),
session_service=session_service,
memory_service=InMemoryMemoryService(),
)
async def execute(self, context: RequestContext, event_queue: EventQueue) -> None:
if self.agent is None:
self._init_agent()
query = context.get_user_input()
updater = TaskUpdater(event_queue, context.task_id, context.context_id)
user_id = context.message.metadata.get("user_id", "a2a-user") if context.message.metadata else "a2a-user"
if not context.current_task:
await updater.submit()
await updater.start_work()
try:
session = await self._get_or_create_session(context.context_id, user_id)
content = types.Content(role="user", parts=[types.Part(text=query)])
async for event in self.runner.run_async(
session_id=session.id, user_id=user_id, new_message=content,
):
if event.is_final_response():
parts = event.content.parts
answer = " ".join(p.text for p in parts if p.text) or "No response."
await updater.add_artifact([TextPart(text=answer)], name="answer")
await updater.complete()
break
except Exception as e:
await updater.update_status(
TaskState.failed, message=new_agent_text_message(f"Error: {e!s}"),
)
raise
async def _get_or_create_session(self, context_id: str, user_id: str):
app_name = self.runner.app_name
if context_id:
session = await self.runner.session_service.get_session(
app_name=app_name, session_id=context_id, user_id=user_id,
)
if session:
return session
session = await self.runner.session_service.create_session(
app_name=app_name, user_id=user_id, session_id=context_id,
)
return session
async def cancel(self, context: RequestContext, event_queue: EventQueue) -> NoReturn:
raise ServerError(error=UnsupportedOperationError())
実行プログラムは環境を自動的に検出します。GOOGLE_CLOUD_AGENT_ENGINE_ID が設定されている場合（Agent Runtime はデプロイ時にこれを挿入します）、永続セッションに VertexAiSessionService を使用します。ローカルでは、InMemorySessionService にフォールバックします。
reservation_agent ディレクトリには次のものが含まれます。
reservation_agent/ ├── __init__.py ├── agent.py ├── a2a_config.py ├── executor.py └── .env
7. Agent Platform SDK を使用して A2A エージェントを準備し、ローカルでテストする
このステップでは、Agent Platform SDK（下位互換性のため、SDK 名では vertex という用語が引き続き使用されています）の A2aAgent クラスを使用して予約エージェントを A2A 準拠のエージェントとしてラップし、エージェント カードの取得、メッセージの送信、タスクの取得など、A2A プロトコルのフロー全体をローカルでテストします。これは、次のステップで Agent ランタイムにデプロイする A2aAgent オブジェクトと同じです。
依存関係を追加する
Agent ランタイムと ADK のサポートに加えて A2A SDK を含む Agent Platform SDK をインストールします。
uv add "google-cloud-aiplatform[agent_engines,adk]==1.149.0" "a2a-sdk==0.3.26"
A2A コンポーネントについて
A2A 用の ADK エージェントをラップするには、次の 3 つのコンポーネントが必要です。
- エージェント カード - エージェントの機能、スキル、エンドポイント URL を記述した「名刺」です。他のエージェントは、これを使用してエージェントの動作を検出します。
- エージェント エグゼキュータ - A2A プロトコルと ADK エージェントのロジック間のブリッジ。A2A リクエストを受け取り、ADK エージェントを介して実行し、結果を A2A タスクとして返します。
- A2aAgent - カードとエグゼキュータをデプロイ可能な単位に結合する Agent Platform SDK クラス。
テスト スクリプトを作成する
ローカルでテストするスクリプトを作成する
cloudshell edit scripts/test_a2a_agent_local.py
以下を scripts/test_a2a_agent_local.py にコピーします。
# scripts/test_a2a_agent_local.py
import asyncio
import json
import os
from pprint import pprint
from dotenv import load_dotenv
from starlette.requests import Request
from vertexai.preview.reasoning_engines import A2aAgent
from reservation_agent.a2a_config import agent_card
from reservation_agent.executor import ReservationAgentExecutor
load_dotenv()
# --- Helper functions for building mock requests ---
def receive_wrapper(data: dict):
async def receive():
byte_data = json.dumps(data).encode("utf-8")
return {"type": "http.request", "body": byte_data, "more_body": False}
return receive
def build_post_request(data: dict = None, path_params: dict = None) -> Request:
scope = {"type": "http", "http_version": "1.1", "headers": [(b"content-type", b"application/json")], "app": None}
if path_params:
scope["path_params"] = path_params
return Request(scope, receive_wrapper(data))
def build_get_request(path_params: dict) -> Request:
scope = {"type": "http", "http_version": "1.1", "query_string": b"", "app": None}
if path_params:
scope["path_params"] = path_params
async def receive():
return {"type": "http.disconnect"}
return Request(scope, receive)
# --- Helper: poll for task completion ---
async def wait_for_task(a2a_agent, task_id, max_retries=30):
"""Poll on_get_task until the task reaches a terminal state."""
for _ in range(max_retries):
request = build_get_request({"id": task_id})
result = await a2a_agent.on_get_task(request=request, context=None)
state = result.get("status", {}).get("state", "")
if state in ["completed", "failed"]:
return result
await asyncio.sleep(1)
return result
def print_task_answer(result):
"""Extract and print the answer from task artifacts."""
print(f"Status: {result.get('status', {}).get('state')}")
for artifact in result.get("artifacts", []):
if artifact.get("parts") and "text" in artifact["parts"][0]:
print(f"Answer: {artifact['parts'][0]['text']}")
# --- Local test ---
async def main():
# Create and set up the A2A agent locally
a2a_agent = A2aAgent(agent_card=agent_card, agent_executor_builder=ReservationAgentExecutor)
a2a_agent.set_up()
# 1. Get agent card
print("=" * 50)
print("1. Retrieving agent card...")
print("=" * 50)
request = build_get_request(None)
card_response = await a2a_agent.handle_authenticated_agent_card(request=request, context=None)
print(f"Agent: {card_response.get('name')}")
print(f"Skills: {[s.get('name') for s in card_response.get('skills', [])]}")
# 2. Create a reservation
print("\n" + "=" * 50)
print("2. Creating a reservation...")
print("=" * 50)
message_data = {
"message": {
"messageId": f"msg-{os.urandom(4).hex()}",
"content": [{"text": "Book a table for 2 on Saturday at 6pm. Name: Bob, Phone: 555-0202"}],
"role": "ROLE_USER",
},
}
request = build_post_request(message_data)
response = await a2a_agent.on_message_send(request=request, context=None)
task_id = response["task"]["id"]
context_id = response["task"].get("contextId")
print(f"Task ID: {task_id}")
# 3. Wait for result
print("\n" + "=" * 50)
print("3. Waiting for task result...")
print("=" * 50)
result = await wait_for_task(a2a_agent, task_id)
print_task_answer(result)
# 4. Check the reservation (same context for session continuity)
print("\n" + "=" * 50)
print("4. Checking the reservation...")
print("=" * 50)
check_data = {
"message": {
"messageId": f"msg-{os.urandom(4).hex()}",
"content": [{"text": "Check the reservation for 555-0202"}],
"role": "ROLE_USER",
"contextId": context_id,
},
}
request = build_post_request(check_data)
check_response = await a2a_agent.on_message_send(request=request, context=None)
check_result = await wait_for_task(a2a_agent, check_response["task"]["id"])
print_task_answer(check_result)
# 5. Cancel the reservation
print("\n" + "=" * 50)
print("5. Cancelling the reservation...")
print("=" * 50)
cancel_data = {
"message": {
"messageId": f"msg-{os.urandom(4).hex()}",
"content": [{"text": "Cancel the reservation for 555-0202"}],
"role": "ROLE_USER",
"contextId": context_id,
},
}
request = build_post_request(cancel_data)
cancel_response = await a2a_agent.on_message_send(request=request, context=None)
cancel_result = await wait_for_task(a2a_agent, cancel_response["task"]["id"])
print_task_answer(cancel_result)
print("\n" + "=" * 50)
print("All tests passed!")
print("=" * 50)
if __name__ == "__main__":
asyncio.run(main())
テストスクリプトは、前の手順で作成したエージェントカードとエグゼキュータをインポートします。重複はありません。ローカル A2aAgent を作成し、モック HTTP リクエストを介して A2A プロトコル呼び出しをシミュレートし、3 つの予約オペレーションすべてを検証します。
ローカルに GOOGLE_CLOUD_AGENT_ENGINE_ID が設定されていないため、エグゼキュータは InMemorySessionService を使用します。Agent Runtime にデプロイすると、同じエグゼキュータが永続セッション用に VertexAiSessionService に自動的に切り替わります。
テストを実行する
PYTHONPATH=. uv run python scripts/test_a2a_agent_local.py
出力には、次の 5 つのステージが表示されます。
- エージェント カード - エージェントの機能とスキルを取得します。
- 予約を作成 - テーブルを予約し、確認を含むタスクを返します
- タスクの結果を取得 - 回答を含む完了したタスクを取得します。
- Check reservation（予約を確認） - 電話番号で予約を検索します
- Cancel reservation（予約をキャンセル） - 予約をキャンセルして確定します。
出力例（以下を参照）
================================================== 1. Retrieving agent card... ================================================== Agent: Reservation Agent Skills: ['Restaurant Reservations'] ================================================== 2. Creating a reservation... ================================================== Task ID: f7f7004d-cfea-49c2-b57d-5bca9959e193 ================================================== 3. Waiting for task result... ================================================== Status: TASK_STATE_COMPLETED Answer: Your reservation for Bob, party of 2, on Saturday at 6:00 PM has been confirmed. The phone number associated is 555-0202. ================================================== 4. Checking the reservation... ================================================== Status: TASK_STATE_COMPLETED Answer: I found a reservation for Bob, party of 2, on Saturday at 6:00 PM. The reservation status is confirmed. ================================================== 5. Cancelling the reservation... ================================================== Status: TASK_STATE_COMPLETED Answer: Your reservation for Bob (555-0202) has been cancelled. ================================================== All tests passed! ==================================================
この時点で、A2A エージェント カードに正しいスキルが記述されていること、3 つの予約オペレーションすべてが A2A プロトコルのメッセージ/タスクフローで動作すること、同じコンテキスト内のメッセージ間で状態が保持されることを確認しました。
8. 予約エージェントを Agent Runtime にデプロイする
このステップでは、予約エージェントを Gemini Enterprise Agent Platform Runtime にデプロイします。これは、エージェントをホストし、安全な A2A エンドポイントとして公開するフルマネージドのサーバーレス プラットフォームです。デプロイ後、承認されたクライアントは標準の A2A HTTP エンドポイントを介してエージェントを検出して操作できます。
ステージング バケットを作成する
Agent Runtime ステージング用の Cloud Storage バケットを作成します。Agent ランタイムは、このバケットを使用して、デプロイ中にエージェントのコードと依存関係をアップロードします。
STAGING_BUCKET="${GOOGLE_CLOUD_PROJECT}-adk-a2a-agent-runtime"
gsutil mb -l $REGION -p $GOOGLE_CLOUD_PROJECT gs://$STAGING_BUCKET 2>/dev/null || echo "Bucket already exists"
echo "STAGING_BUCKET=$STAGING_BUCKET" >> .env
source .env
デプロイ スクリプトを作成する
次に、デプロイ スクリプトを準備する必要があります。
cloudshell edit scripts/deploy_a2a_agent_runtime.py
以下を scripts/deploy_a2a_agent_runtime.py にコピーします。
# scripts/deploy_a2a_agent_runtime.py
import os
from pathlib import Path
import vertexai
from dotenv import load_dotenv
from google.genai import types
from vertexai.preview.reasoning_engines import A2aAgent
from reservation_agent.a2a_config import agent_card
from reservation_agent.executor import ReservationAgentExecutor
load_dotenv()
PROJECT_ID = os.environ["GOOGLE_CLOUD_PROJECT"]
REGION = os.environ["REGION"]
STAGING_BUCKET = os.environ.get("STAGING_BUCKET", f"{PROJECT_ID}-adk-a2a-agent-runtime")
BUCKET_URI = f"gs://{STAGING_BUCKET}"
a2a_agent = A2aAgent(
agent_card=agent_card,
agent_executor_builder=ReservationAgentExecutor,
)
def main():
vertexai.init(project=PROJECT_ID, location=REGION, staging_bucket=BUCKET_URI)
client = vertexai.Client(
project=PROJECT_ID,
location=REGION,
http_options=types.HttpOptions(api_version="v1beta1"),
)
print("Deploying Reservation Agent to Agent Runtime...")
print("This may take 3-5 minutes.")
remote_agent = client.agent_engines.create(
agent=a2a_agent,
config={
"display_name": agent_card.name,
"description": agent_card.description,
"requirements": [
"google-cloud-aiplatform[agent_engines,adk]==1.149.0",
"a2a-sdk==0.3.26",
"google-adk==1.29.0",
"cloudpickle",
"pydantic"
],
"extra_packages": [
"./reservation_agent",
],
"http_options": {
"api_version": "v1beta1",
},
"staging_bucket": BUCKET_URI,
},
)
resource_name = remote_agent.api_resource.name
print(f"\nDeployment complete!")
print(f"Resource name: {resource_name}")
env_path = Path(".env")
lines = env_path.read_text().splitlines() if env_path.exists() else []
lines = [l for l in lines if not l.startswith("RESERVATION_AGENT_RESOURCE_NAME=")]
lines.append(f"RESERVATION_AGENT_RESOURCE_NAME={resource_name}")
env_path.write_text("\n".join(lines) + "\n")
print("Written RESERVATION_AGENT_RESOURCE_NAME to .env")
if __name__ == "__main__":
main()
デプロイ スクリプトは、ローカル テストで使用される同じ agent_card と ReservationAgentExecutor をインポートします。コードの重複はありません。エージェント ランタイムは、デプロイ用に A2aAgent オブジェクトとその依存関係をシリアル化（ピクルス化）します。デプロイ スクリプトの最後に、RESERVATION_AGENT_RESOURCE_NAME の値が .env ファイルに書き込まれます。
Agent Runtime にデプロイする
次のようにデプロイ スクリプトを実行します。
PYTHONPATH=. uv run python scripts/deploy_a2a_agent_runtime.py
デプロイには 3 ～ 5 分かかります。このスクリプトは、予約エージェントをホストするサーバーレス エンドポイントを Agent Runtime にプロビジョニングします。デプロイが成功すると、次のような出力が表示されます。
Deploying Reservation Agent to Agent Runtime... This may take 3-5 minutes. Deployment complete! Resource name: projects/your-project-number/locations/us-central1/reasoningEngines/your-agent-deployment-unique-id Written RESERVATION_AGENT_RESOURCE_NAME to .env
デプロイされたエージェントは、Cloud コンソールで確認できます。コンソールの検索バーで Agent Platform を検索する
次に、左側のタブで Agents にカーソルを合わせ、Deployments を選択します。
次のように、デプロイ リストに Reservation Agent が表示されます。
デプロイしたエージェントをテストする
これで、デプロイされたエージェントをテストする準備が整いました。デプロイされたエージェントのテストスクリプトを作成します。
cloudshell edit scripts/test_a2a_agent_runtime.py
以下を scripts/test_a2a_agent_runtime.py にコピーします。
# scripts/test_a2a_agent_runtime.py
import asyncio
import os
import time
import vertexai
from a2a.types import TaskState
from dotenv import load_dotenv
from google.genai import types
load_dotenv()
PROJECT_ID = os.environ["GOOGLE_CLOUD_PROJECT"]
REGION = os.environ["REGION"]
RESOURCE_NAME = os.environ["RESERVATION_AGENT_RESOURCE_NAME"]
async def main():
vertexai.init(project=PROJECT_ID, location=REGION)
client = vertexai.Client(
project=PROJECT_ID, location=REGION,
http_options=types.HttpOptions(api_version="v1beta1"),
)
agent = client.agent_engines.get(name=RESOURCE_NAME)
# 1. Get agent card
print("=" * 50)
print("1. Retrieving agent card...")
print("=" * 50)
card = await agent.handle_authenticated_agent_card()
print(f"Agent: {card.name}")
print(f"URL: {card.url}")
print(f"Skills: {[s.name for s in card.skills]}")
# 2. Send a reservation request
print("\n" + "=" * 50)
print("2. Sending reservation request...")
print("=" * 50)
message_data = {
"messageId": "msg-remote-001",
"role": "user",
"parts": [{"kind": "text", "text": "Book a table for 3 on Sunday at noon. Name: Carol, Phone: 555-0303"}],
}
response = await agent.on_message_send(**message_data)
task_object = None
for chunk in response:
if isinstance(chunk, tuple) and len(chunk) > 0 and hasattr(chunk[0], "id"):
task_object = chunk[0]
break
task_id = task_object.id
print(f"Task ID: {task_id}")
print(f"Status: {task_object.status.state}")
# 3. Poll for result
print("\n" + "=" * 50)
print("3. Waiting for result...")
print("=" * 50)
result = None
for _ in range(30):
try:
result = await agent.on_get_task(id=task_id)
if result.status.state in [TaskState.completed, TaskState.failed]:
break
except Exception:
pass
time.sleep(1)
print(f"Final status: {result.status.state}")
if result.artifacts:
for artifact in result.artifacts:
if artifact.parts and hasattr(artifact.parts[0], "root") and hasattr(artifact.parts[0].root, "text"):
print(f"Answer: {artifact.parts[0].root.text}")
print("\n" + "=" * 50)
print("Remote agent test passed!")
print("=" * 50)
if __name__ == "__main__":
asyncio.run(main())
では、テストを実行してみましょう。
source .env
uv run python scripts/test_a2a_agent_runtime.py
出力には、[Restaurant Reservations] スキルを持つエージェント カードが表示され、その後に予約確認でタスクが完了します。
================================================== 1. Retrieving agent card... ================================================== Agent: Reservation Agent URL: https://us-central1-aiplatform.googleapis.com/v1beta1/projects/your-project-id/locations/us-central1/reasoningEngines/your-agent-unique-id/a2a Skills: ['Restaurant Reservations'] ================================================== 2. Sending reservation request... ================================================== Task ID: b34585d0-5f03-4cb0-85a3-40710a0d224d Status: TaskState.completed ================================================== 3. Waiting for result... ================================================== Final status: TaskState.completed Answer: Your reservation for Carol, party of 3 on Sunday at noon with phone number 555-0303 is confirmed. ================================================== Remote agent test passed! ==================================================
予約エージェントが、Agent Runtime でマネージド A2A エンドポイントとして正常に実行されるようになりました。
9. A2A 予約エージェントをルート レストラン エージェントと統合する
このステップでは、デプロイされた予約エージェントをリモート A2A サブエージェントとして使用するようにレストラン エージェントをアップグレードします。オーケストレーターはローカルで実行され、予約エージェントは Agent Runtime で実行されます。これは、完全なデプロイの前に A2A 接続を検証する部分的な統合です。
A2A エージェント カードの URL を解決する
RemoteA2aAgent は、デプロイされた予約エージェントのカード URL を使用して、その機能を検出する必要があります。この URL をエージェント ランタイムから取得し、レストラン エージェントの .env に書き込むスクリプトを作成します。
cloudshell edit scripts/resolve_agent_card_url.py
以下を scripts/resolve_agent_card_url.py にコピーします。
# scripts/resolve_agent_card_url.py
import asyncio
import os
from pathlib import Path
import vertexai
from dotenv import load_dotenv
from google.genai import types
load_dotenv()
PROJECT_ID = os.environ["GOOGLE_CLOUD_PROJECT"]
REGION = os.environ["REGION"]
RESOURCE_NAME = os.environ["RESERVATION_AGENT_RESOURCE_NAME"]
async def main():
vertexai.init(project=PROJECT_ID, location=REGION)
client = vertexai.Client(
project=PROJECT_ID, location=REGION,
http_options=types.HttpOptions(api_version="v1beta1"),
)
agent = client.agent_engines.get(name=RESOURCE_NAME)
card = await agent.handle_authenticated_agent_card()
card_url = f"{card.url}/v1/card"
print(f"Agent: {card.name}")
print(f"Card URL: {card_url}")
# Write to restaurant_agent/.env
# Write to both restaurant_agent/.env (for adk web) and root .env (for Cloud Run deploy)
for env_path in [Path("restaurant_agent/.env"), Path(".env")]:
lines = env_path.read_text().splitlines() if env_path.exists() else []
lines = [l for l in lines if not l.startswith("RESERVATION_AGENT_CARD_URL=")]
lines.append(f"RESERVATION_AGENT_CARD_URL={card_url}")
env_path.write_text("\n".join(lines) + "\n")
print(f"Written RESERVATION_AGENT_CARD_URL to {env_path}")
if __name__ == "__main__":
asyncio.run(main())
スクリプトを実行して、エージェント カードの URL を .env ファイルに入力します。
uv run python scripts/resolve_agent_card_url.py
source .env
レストラン エージェントを更新する
レストラン エージェント ファイルを開きます。
cloudshell edit restaurant_agent/agent.py
次に、リモート予約エージェントをサブエージェントとして含む更新バージョンに内容を置き換えます。
# restaurant_agent/agent.py
import os
import httpx
from google.adk.agents import LlmAgent
from google.adk.agents.remote_a2a_agent import RemoteA2aAgent
from google.auth import default
from google.auth.transport.requests import Request as AuthRequest
from toolbox_adk import ToolboxToolset
TOOLBOX_URL = os.environ.get("TOOLBOX_URL", "http://127.0.0.1:5000")
RESERVATION_AGENT_CARD_URL = os.environ.get("RESERVATION_AGENT_CARD_URL", "")
toolbox = ToolboxToolset(TOOLBOX_URL)
class GoogleCloudAuth(httpx.Auth):
"""Auto-refreshing Google Cloud authentication for httpx.
Refreshes the access token before each request if expired,
so long-running agents never hit 401 errors.
"""
def __init__(self):
self.credentials, _ = default(
scopes=["https://www.googleapis.com/auth/cloud-platform"]
)
def auth_flow(self, request):
# Refresh the token if it is expired or missing
if not self.credentials.valid:
self.credentials.refresh(AuthRequest())
request.headers["Authorization"] = f"Bearer {self.credentials.token}"
yield request
reservation_remote_agent = RemoteA2aAgent(
name="reservation_agent",
description="Handles restaurant table reservations — create, check, and cancel bookings. Delegate to this agent when the user wants to book a table, check a reservation, or cancel a reservation.",
agent_card=RESERVATION_AGENT_CARD_URL,
httpx_client=httpx.AsyncClient(auth=GoogleCloudAuth(), timeout=60),
)
root_agent = LlmAgent(
name="restaurant_agent",
model="gemini-2.5-flash",
instruction="""You are a friendly and knowledgeable concierge at "Foodie Finds," a restaurant. Your job:
- Help diners browse the menu by category or cuisine type.
- Provide full details about specific dishes, including ingredients, price, and dietary information.
- Recommend dishes based on natural language descriptions of what the diner is craving.
- Add new menu items when asked.
- For reservation requests (booking, checking, or cancelling tables), delegate to the reservation_agent.
When a diner asks about a specific dish by name or cuisine, use the get-item-details tool.
When a diner asks for a specific category or cuisine type, use the search-menu tool.
When a diner describes what kind of food they want — by flavor, texture, dietary needs, or cravings — use the search-menu-by-description tool for semantic search.
When in doubt between search-menu and search-menu-by-description, prefer search-menu-by-description — it searches dish descriptions and finds more relevant matches.
If a dish is not available (available is false), let the diner know and suggest similar alternatives from the search results.
Be conversational, knowledgeable, and concise.""",
tools=[toolbox],
sub_agents=[reservation_remote_agent],
)
以前のバージョンからの主な変更点は次のとおりです。
GoogleCloudAuth- 各リクエストの前に Google Cloud アクセス トークンを更新するカスタムhttpx.Authハンドラ。Agent Runtime では認証済みの A2A 呼び出しが必要であり、トークンは一定期間後に期限切れになります。RemoteA2aAgentは.env（解決スクリプトによって書き込まれたもの）からRESERVATION_AGENT_CARD_URLを読み取り、認証済みのhttpx_clientを使用します。- サブエージェントとして登録されている - ADK のオーケストレーターが予約リクエストを自動的に委任する
- 予約の委任に関する説明を追加しました
統合されたエージェントをローカルでテストする
スターター エージェントには MCP ツールボックスとの統合が必要であり、必要なファイルは以前の Codelab またはスターター リポジトリからすでに提供されているはずです。ツールボックス プロセスが適切に実行されるようにするだけで済みます。
.env の TOOLBOX_URL がすでに Cloud Run サービス（前の Codelab またはスターター リポジトリの full_setup.sh）を指している場合は、この手順をスキップできます。エージェントはデプロイされた Toolbox に接続します。
ローカルの Toolbox が必要な場合は、新しいインスタンスを開始する前に、すでに実行されているかどうかを確認します。
if curl -s http://127.0.0.1:5000/api/toolsets > /dev/null 2>&1; then
echo "Toolbox already running on port 5000"
else
set -a; source .env; set +a
./toolbox --config=tools.yaml > logs/toolbox.log 2>&1 &
echo "Toolbox started"
fi
次に、ADK ウェブ開発 UI を介してレストラン エージェントを操作してみましょう。
uv run adk web --allow_origins "regex:https://.*\.cloudshell\.dev" --port 8080
Cloud Shell ウェブ プレビューを使用して ADK ウェブ UI を開き（[ウェブでプレビュー] ボタンをクリックし、ポートを 8080 に変更）、restaurant_agent を選択します。
混合会話をテストする:
メニュー クエリ
What Italian dishes do you have?
予約リクエスト
I want to create reservation under name Bob, phone number 123456
予約を確認する
新しいセッションを作成する（新しい会話を開始する）:
Check the reservation for 123456
Ctrl+C キーを 2 回押して、adk web プロセスを停止します。次に、エージェントを完全にデプロイしてシステムを完成させます。
10. 更新されたレストラン エージェントを Cloud Run にデプロイする
この手順では、A2A 統合を使用してレストラン エージェントを Cloud Run に再デプロイし、完全にデプロイされたマルチエージェント システムを完成させます。
Agent Runtime へのアクセス権限を付与する
Cloud Run サービス アカウントには、エージェント ランタイムを呼び出す権限が必要です。デフォルトの Compute Engine サービス アカウントに roles/aiplatform.user ロールを付与します。
PROJECT_NUMBER=$(gcloud projects describe $GOOGLE_CLOUD_PROJECT --format='value(projectNumber)')
gcloud projects add-iam-policy-binding $GOOGLE_CLOUD_PROJECT \
--member="serviceAccount:${PROJECT_NUMBER}-compute@developer.gserviceaccount.com" \
--role="roles/aiplatform.user"
Cloud Run へのデプロイ
この設定では、前の Codelab で作成したレストラン エージェント サービスがすでに存在しているか、新たに開始する場合は scripts/full_setup.sh を実行していることを前提としています。これにより、更新されたコード（新しい RemoteA2aAgent 統合）で再デプロイされ、予約エージェント カードの URL が新しい環境変数として追加されます。既存の環境変数（TOOLBOX_URL、GOOGLE_CLOUD_PROJECT など）は保持されます。
gcloud run deploy restaurant-agent \
--source . \
--region=$REGION \
--allow-unauthenticated \
--update-env-vars="RESERVATION_AGENT_CARD_URL=$RESERVATION_AGENT_CARD_URL" \
--min-instances=0 \
--max-instances=1 \
--memory=1Gi \
--port=8080
完全にデプロイされたシステムをテストする
デプロイされたサービスの URL を取得します。
AGENT_URL=$(gcloud run services describe restaurant-agent --region=$REGION --format='value(status.url)')
echo "Agent URL: $AGENT_URL"
ブラウザでその URL を開きます。ADK ウェブ UI が読み込まれます。これは、ローカルで使用したのと同じインターフェースで、Cloud Run で実行されています。
エージェントと雑談しても構いません
メニュー クエリ
What spicy dishes do you have?
予約リクエスト
Book a table for 4 on Friday at 7pm. Name: Eve, Phone: 555-0505
予約を確認する
新しいセッションを作成する（新しい会話を開始する）:
Check reservation for 555-0505
マルチエージェント システムが完全にデプロイされている。Cloud Run のレストラン エージェントは、メニュー オペレーション用の MCP Toolbox と Agent Runtime の A2A 予約エージェントという 2 つのバックエンド サービスをオーケストレートします。
11. 完了
Google Cloud で A2A プロトコルを使用してマルチエージェント システムを構築してデプロイした。
学習した内容
- データベースを使用せずにセッション状態（
ToolContext）を使用して予約データを管理する ADK エージェントを構築しました - Agent Platform SDK を使用して A2A エージェントを Agent ランタイムにデプロイした
RemoteA2aAgentをサブエージェントとして使用して、別の ADK エージェントからリモート A2A エージェントを消費した- システムを段階的にテストしました（ローカル A2A → デプロイされた A2A → 部分的な統合 → 完全なデプロイ）。
クリーンアップ
Google Cloud アカウントに課金されないようにするには、この Codelab で作成したリソースを削除します。
オプション 1: プロジェクトを削除する（推奨）
gcloud projects delete $GOOGLE_CLOUD_PROJECT
オプション 2: 個々のリソースを削除する
# Delete the Agent Runtime deployment
uv run python -c "
import vertexai
from google.genai import types
vertexai.init(project='$GOOGLE_CLOUD_PROJECT', location='$REGION')
client = vertexai.Client(
project='$GOOGLE_CLOUD_PROJECT', location='$REGION',
http_options=types.HttpOptions(api_version='v1beta1'),
)
agent = client.agent_engines.get(name='$RESERVATION_AGENT_RESOURCE_NAME')
agent.delete(force=True)
print('Agent Runtime deployment deleted.')
"
# Delete Cloud Run services
gcloud run services delete restaurant-agent --region=$REGION --quiet
gcloud run services delete toolbox-service --region=$REGION --quiet
# Delete Cloud SQL instance
gcloud sql instances delete $DB_INSTANCE --quiet
# Delete GCS staging bucket
gsutil rm -r gs://$STAGING_BUCKET