Google uses AI technology to translate content into your preferred language. AI translations can contain errors.

大規模なエージェント: エージェントランタイムと ADK 統合での A2A プロトコルを使用したマルチエージェントアーキテクチャ

1. はじめに

AI エージェントがより多くの責任を負うようになると、1 つのエージェントがすべてを行うことは、維持、スケーリング、進化が困難になります。多くの場合、機能ごとに異なるデプロイ戦略、更新サイクル、さらには異なるチームによる所有が必要になります。

A2A（Agent2Agent）プロトコルは、通信側の問題を解決します。つまり、エージェントが互いの機能を検出し、フレームワークや組織を越えて連携する方法を標準化します。
Gemini Enterprise Agent Platform Runtime は、デプロイ側の問題を解決します。これは、A2A サポート、自動スケーリング、安全なエンドポイント、永続セッション、インフラストラクチャ管理の不要なエージェントをホストするフルマネージドのサーバーレスプラットフォームです。

これらを組み合わせることで、特化型エージェントを構築し、検出可能な A2A サービスとしてデプロイして、マルチエージェントシステムに構成できます。

作成するアプリの概要

Gemini Enterprise Agent Platform Sessions によって管理される ADK セッション状態を使用して、レストランのテーブル予約（作成、確認、キャンセル）を管理する予約エージェント。このエージェントを Gemini Enterprise Agent Platform Runtime にデプロイすると、A2A プロトコルのエージェントカードで検出できるようになります。次に、Foodie Finds レストランコンシェルジュエージェント（前提条件の Codelab のもの。Codelab をまだご覧になっていない場合でも、スターターリポジトリをご用意していますのでご安心ください）をアップグレードして、予約エージェントをリモート A2A サブエージェントとして使用します。その結果、オーケストレーターがメニュークエリを MCP ツールボックスに、予約リクエストをリモート A2A エージェントにルーティングするマルチエージェントシステムが実現します。

学習内容

マネージドセッションサービスを使用して予約データを管理する ADK エージェントを構築する
エージェントカードとスキルを使用して ADK エージェントを A2A サーバーとして公開する
A2A エージェントを Gemini Enterprise Agent Runtime にデプロイする
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 に必要なすべてのツールがプリインストールされた環境が用意されています。プロンプトが表示されたら、[承認] をクリックします。

[表示] -> [ターミナル] をクリックしてターミナルを開きます。インターフェースは次のようになります。

これがメインインターフェースになります。上部に 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 にデプロイします。デプロイプロセスでは、エージェントコードをシリアル化（ピクルス化）してアップロードします。Agent Runtime は、A2A プロトコルを処理するサーバーレスエンドポイントをプロビジョニングします。他のエージェント（またはクライアント）は、Google Cloud 認証情報で認証された標準の HTTP 呼び出しを介してこのエンドポイントとやり取りします。

5. 予約エージェントを構築する

このステップでは、セッション状態を使用してレストランの予約を処理する新しい ADK エージェントを作成します。エージェントは、電話番号をルックアップキーとして、作成、確認、キャンセルという 3 つのオペレーションをサポートしています。すべての予約データは ADK のセッション状態に保存されます

エージェントをスキャフォールディングする

adk create を使用して、正しいモデルとプロジェクト構成でエージェントディレクトリ構造を生成します。

source .env
uv run adk create reservation_agent \
    --model gemini-3.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
import os
from functools import cached_property
from typing import Any

from google.adk.agents import LlmAgent
from google.adk.models.google_llm import Gemini
from google.adk.tools import ToolContext
from google.genai import Client, types

# App-scoped state prefix ensures reservations persist across all sessions.
# See https://adk.dev/sessions/state/ for state scope details.
STATE_PREFIX = "app:reservation:"


class GeminiGlobal(Gemini):
    """Gemini with location pinned to 'global'.

    A2aAgent.set_up() on Agent Platform Runtime SDK overwrites GOOGLE_CLOUD_LOCATION with the Agent Platform Runtime
    region (e.g. us-central1), but the Gemini 3-family endpoint requires 'global'.
    """

    @cached_property
    def api_client(self) -> Client:
        project = os.getenv("GOOGLE_CLOUD_PROJECT")
        return Client(
            project=project,
            location="global",
            http_options=types.HttpOptions(
                headers=self._tracking_headers(),
                retry_options=self.retry_options,
            ),
        )


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=GeminiGlobal(model="gemini-3.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 Runtime にデプロイする A2aAgent オブジェクトと同じです。

依存関係を追加する

Agent Runtime と 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 ランタイムにデプロイします。これは、エージェントをホストし、安全な 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 をインポートします。コードの重複はありません。Agent Runtime は、デプロイ用に A2aAgent オブジェクトとその依存関係をシリアル化（ピクルス化）します。デプロイスクリプトの最後に、RESERVATION_AGENT_RESOURCE_NAME の値が .env ファイルに書き込まれます。

Agent Runtime にデプロイする

次のようにデプロイスクリプトを実行します。

PYTHONPATH=. uv run python scripts/deploy_a2a_agent_runtime.py

デプロイには 3 ～ 5 分かかります。スクリプトは、予約エージェントをホストするサーバーレスエンドポイントをエージェントランタイムにプロビジョニングします。デプロイが成功すると、次のような出力が表示されます。

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 サブエージェントとして使用するように、レストランエージェントをアップグレードします。オーケストレーターはローカルで実行され、予約エージェントはエージェントランタイムで実行されます。これは、完全なデプロイの前に 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-3.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 Runtime にデプロイした
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

大規模なエージェント: エージェント ランタイムと ADK 統合での A2A プロトコルを使用したマルチエージェント アーキテクチャ

1. はじめに

作成するアプリの概要

学習内容

前提条件

2. 環境のセットアップ - 前の Codelab からの継続

必要な API を有効にする

3. 環境のセットアップ - スターター リポジトリで新たに開始する

Cloud Shell を開く

作業ディレクトリを設定する

必要な API を有効にする

スターター インフラストラクチャの設定

4. コンセプト: Agent2Agent（A2A）プロトコルと Gemini Enterprise エージェント ランタイム

Agent2Agent（A2A）プロトコル

Gemini Enterprise Agent Platform Runtime

5. 予約エージェントを構築する

エージェントをスキャフォールディングする

エージェント コードを記述する

6. A2A サーバー構成を準備する

A2A エージェント カードを定義する

A2A エグゼキュータを作成する

7. Agent Platform SDK を使用して A2A エージェントを準備し、ローカルでテストする

依存関係を追加する

A2A コンポーネントについて

テスト スクリプトを作成する

テストを実行する

8. 予約エージェントを Agent Runtime にデプロイする

ステージング バケットを作成する

デプロイ スクリプトを作成する

Agent Runtime にデプロイする

デプロイしたエージェントをテストする

9. A2A 予約エージェントをルート レストラン エージェントと統合する

A2A エージェント カードの URL を解決する

レストラン エージェントを更新する

統合エージェントをローカルでテストする

10. 更新されたレストラン エージェントを Cloud Run にデプロイする

Agent Runtime へのアクセス権を付与する

Cloud Run へのデプロイ

完全にデプロイされたシステムをテストする

11. 完了

学習した内容

クリーンアップ

オプション 1: プロジェクトを削除する（推奨）

オプション 2: 個々のリソースを削除する

大規模なエージェント: エージェントランタイムと ADK 統合での A2A プロトコルを使用したマルチエージェントアーキテクチャ

3. 環境のセットアップ - スターターリポジトリで新たに開始する

スターターインフラストラクチャの設定

4. コンセプト: Agent2Agent（A2A）プロトコルと Gemini Enterprise エージェントランタイム

エージェントコードを記述する

A2A エージェントカードを定義する

テストスクリプトを作成する

ステージングバケットを作成する

デプロイスクリプトを作成する

9. A2A 予約エージェントをルートレストランエージェントと統合する

A2A エージェントカードの URL を解決する

レストランエージェントを更新する

10. 更新されたレストランエージェントを Cloud Run にデプロイする