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

Agents at Scale: Multi-Agent Architecture with A2A Protocol on Agent Runtime and ADK Integration

1. Einführung

Da KI-Agenten immer mehr Aufgaben übernehmen, wird es schwierig, einen einzelnen Agenten zu verwalten, zu skalieren und weiterzuentwickeln, der alles erledigt. Für verschiedene Funktionen sind oft unterschiedliche Bereitstellungsstrategien, Aktualisierungszyklen oder sogar unterschiedliche Teams erforderlich.

Das A2A-Protokoll (Agent2Agent) löst das Kommunikationsproblem, indem es standardisiert, wie Agenten die Fähigkeiten des jeweils anderen erkennen und über Frameworks und Organisationen hinweg zusammenarbeiten.
Die Gemini Enterprise Agent Platform Runtime löst das Problem der Bereitstellung. Sie ist eine vollständig verwaltete, serverlose Plattform, auf der Ihre Agents mit integrierter A2A-Unterstützung, automatischer Skalierung, sicheren Endpunkten, persistenten Sitzungen und ohne Infrastrukturverwaltung gehostet werden.

Damit können Sie spezialisierte Agenten erstellen, sie als auffindbare A2A-Dienste bereitstellen und in Systeme mit mehreren Agenten einbinden.

Umfang

Ein Reservierungsagent, der Restaurantreservierungen (Erstellen, Prüfen und Stornieren) über den ADK-Sitzungsstatus verwaltet, der von Gemini Enterprise Agent Platform Sessions verwaltet wird. Sie stellen diesen Agenten in der Gemini Enterprise Agent Platform Runtime bereit, wo er über die Agentenkarte des A2A-Protokolls auffindbar ist. Anschließend aktualisieren Sie den Restaurant-Concierge-Agenten Foodie Finds (aus dem erforderlichen Codelab – keine Sorge, wenn Sie das Codelab noch nicht durchlaufen haben, wir haben ein Starter-Repository für Sie vorbereitet), um den Reservierungs-Agenten als A2A-Remote-Unteragenten zu verwenden. Das Ergebnis ist ein Multi-Agenten-System, in dem der Orchestrator Menüanfragen an die MCP-Toolbox und Reservierungsanfragen an den Remote-A2A-Agenten weiterleitet.

Lerninhalte

ADK-Agenten erstellen, der den verwalteten Sitzungsdienst zum Verwalten von Reservierungsdaten verwendet
ADK-Agent als A2A-Server mit Agentenkarten und Skills bereitstellen
A2A-Agenten in der Gemini Enterprise Agent Runtime bereitstellen
Remote-A2A-Agenten über RemoteA2aAgent in einem anderen ADK-Agenten verwenden und authentifizierte Anfragen verarbeiten
Multi-Agenten-Systeme inkrementell testen: lokales A2A, bereitgestelltes A2A, Teilintegration, vollständige Bereitstellung

Voraussetzungen

(Empfohlen) Sie haben die folgenden Codelabs abgeschlossen :
Persistente KI-Agenten mit ADK und Cloud SQL erstellen –> weitere Informationen zu ADK-Sitzungen und ‑Status
Agentische RAG mit ADK, MCP Toolbox und Cloud SQL: Sie können Ihren Agenten mit diesem Codelab weiterentwickeln. Der bereitgestellte Startcode ist identisch.
Ein Google Cloud-Konto mit einem aktiven Rechnungskonto
Grundlegende Kenntnisse von Python und ADK-Konzepten

2. Umgebung einrichten – Fortsetzung des vorherigen Codelabs

Die in diesem Codelab bereitgestellten Narrative sind die Fortsetzung des erforderlichen Codelabs: Agentic RAG with ADK, MCP Toolbox, and Cloud SQL . Sie können mit dem vorherigen Codelab fortfahren.

Wir können mit dem Erstellen im Arbeitsverzeichnis des vorherigen Codelabs beginnen ( das Arbeitsverzeichnis sollte build-agent-adk-toolbox-cloudsql sein). Um Verwirrung zu vermeiden, benennen wir das Verzeichnis mit demselben Verzeichnisnamen um, den wir verwenden, wenn wir von vorn beginnen.

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

Prüfen Sie, ob die Schlüsseldateien aus dem vorherigen Codelab vorhanden sind:

echo "--- Restaurant Agent ---"
cat restaurant_agent/agent.py | head -5
echo ""
echo "--- Toolbox Config ---"
cat tools.yaml | head -5

Sie sollten die Datei restaurant_agent/agent.py mit dem Import LlmAgent und die Datei tools.yaml mit Ihrer Toolbox-Konfiguration sehen.

Als Nächstes initialisieren wir unsere Python-Umgebung neu.

rm -rf .venv
uv sync

Prüfen Sie außerdem, ob die Datenbank initialisiert und bereit ist:

uv run python scripts/verify_seed.py

Wenn Sie alle Testdetails aus dem vorherigen Codelab befolgen, sehen Sie möglicherweise eine Ausgabe wie diese:

Menu Items: 16/15
Embeddings: 16/15

✗ Database not ready

Das ist kein Problem. Beim Datenbankcheck werden die zusätzlichen Daten, die Sie beim Datenaufnahmecheck eingeben, nicht berücksichtigt. Solange Sie mindestens 15 Daten haben, ist alles in Ordnung.

Erforderliche API aktivieren

Als Nächstes müssen wir die erforderliche API aktivieren, um mit der Gemini Enterprise Agent Platform interagieren zu können.

gcloud services enable \
  cloudresourcemanager.googleapis.com

Sie sollten bereits die erforderlichen Dateien und die Infrastruktur haben, um mit dem nächsten Abschnitt fortzufahren: A2A Protocol and Gemini Enterprise Agent Runtime.

3. Umgebung einrichten – Mit dem Starter-Repository neu beginnen

In diesem Schritt wird Ihre Cloud Shell-Umgebung vorbereitet, Ihr Google Cloud-Projekt konfiguriert und das Starter-Repository geklont.

Cloud Shell öffnen

Öffnen Sie Cloud Shell in Ihrem Browser. Cloud Shell bietet eine vorkonfigurierte Umgebung mit allen Tools, die Sie für dieses Codelab benötigen. Klicken Sie auf Autorisieren, wenn Sie dazu aufgefordert werden.

Klicken Sie dann auf Ansicht -> Terminal, um das Terminal zu öffnen.Die Benutzeroberfläche sollte so aussehen:

Das ist unsere Hauptoberfläche: oben die IDE, unten das Terminal.

Arbeitsverzeichnis einrichten

Klonen Sie das Starter-Repository. Der gesamte Code, den Sie in diesem Codelab schreiben, wird hier gespeichert:

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

Erstellen Sie die Datei .env aus der bereitgestellten Vorlage:

cp .env.example .env

Um die Einrichtung des Projekts in Ihrem Terminal zu vereinfachen, laden Sie dieses Setupscript in Ihr Arbeitsverzeichnis herunter:

curl -sL https://raw.githubusercontent.com/alphinside/cloud-trial-project-setup/main/setup_verify_trial_project.sh -o setup_verify_trial_project.sh

Führen Sie das Skript aus. Dabei wird Ihr Probeabo-Rechnungskonto überprüft, ein neues Projekt erstellt (oder ein vorhandenes Projekt validiert), Ihre Projekt-ID in einer .env-Datei im aktuellen Verzeichnis gespeichert und das aktive Projekt in gcloud festgelegt.

bash setup_verify_trial_project.sh && source .env

Das Skript führt Folgendes aus:

Prüfen, ob Sie ein aktives Testabrechnungskonto haben
Prüfen Sie, ob in .env ein Projekt vorhanden ist.
Neues Projekt erstellen oder vorhandenes Projekt wiederverwenden
Test-Rechnungskonto mit Ihrem Projekt verknüpfen
Speichern Sie die Projekt-ID in .env.
Projekt als aktives gcloud-Projekt festlegen

Prüfen Sie, ob das Projekt richtig eingestellt ist. Sehen Sie dazu im Cloud Shell-Terminal-Prompt nach dem gelben Text neben Ihrem Arbeitsverzeichnis. Dort sollte Ihre Projekt-ID angezeigt werden.

Erforderliche API aktivieren

Als Nächstes müssen wir die erforderliche API aktivieren, um mit der Gemini Enterprise Agent Platform interagieren zu können.

gcloud services enable \
  aiplatform.googleapis.com \
  cloudresourcemanager.googleapis.com

Starter Infrastructure Setup

Zuerst müssen wir Python-Abhängigkeiten mit uv installieren. uv ist ein schneller Python-Paket- und Projektmanager, der in Rust geschrieben wurde ( uv-Dokumentation ). In diesem Codelab wird er verwendet, um das Python-Projekt schnell und einfach zu verwalten.

uv sync

Führen Sie dann das vollständige Setupscript aus, mit dem die Cloud SQL-Instanz erstellt, Daten eingefügt und der Toolbox-Dienst bereitgestellt wird, der als Ausgangszustand unseres Restaurant-Agents dient.

bash scripts/full_setup.sh > logs/full_setup.log 2>&1 &

4. Konzept: Agent2Agent-Protokoll (A2A) und Gemini Enterprise Agent Runtime

Bevor wir mit der Entwicklung beginnen, wollen wir uns kurz die beiden wichtigsten Technologien ansehen, die in diesem Codelab vorgestellt werden, um unsere Agentenanwendung zu skalieren.

Agent2Agent-Protokoll (A2A)

Das Agent2Agent-Protokoll (A2A) ist ein offener Standard, der eine nahtlose Kommunikation und Zusammenarbeit zwischen KI-Agenten ermöglichen soll. Während MCP (Model Context Protocol) Agenten mit Tools und Daten verbindet, verbindet A2A Agenten mit anderen Agenten. So können sie die Funktionen des jeweils anderen erkennen, Aufgaben delegieren und über Frameworks und Organisationen hinweg zusammenarbeiten.

Der Hauptunterschied zwischen dem Einbinden eines Agenten als Tool (über MCP) und dem Bereitstellen über A2A besteht darin, dass Tools zustandslos sind und einzelne Funktionen ausführen, während A2A-Agenten Schlussfolgerungen ziehen, den Status beibehalten und Interaktionen mit mehreren Durchgängen wie Verhandlungen oder Klarstellungen verarbeiten können. Ein über A2A verfügbar gemachter Agent behält seine vollen Funktionen bei und wird nicht auf einen Funktionsaufruf reduziert.

A2A definiert drei Kernkonzepte:

Agentenkarte: Ein JSON-Dokument, das beschreibt, was ein Agent tut, welche Fähigkeiten er hat und welcher Endpunkt verwendet wird. Andere Agenten rufen diese Karte ab, um Funktionen zu ermitteln.
Nachricht: Eine Nutzer- oder Agent-Anfrage, die an einen A2A-Endpunkt gesendet wird und eine Aufgabe auslöst.
Aufgabe: Eine Arbeitseinheit mit einem Lebenszyklus (eingereicht → in Bearbeitung → abgeschlossen/fehlgeschlagen) und Artefakten, die die Ergebnisse enthalten.

Weitere Informationen finden Sie unter Was ist A2A?

Gemini Enterprise Agent Platform Runtime

Agent Runtime ist ein vollständig verwalteter Dienst in Google Cloud, mit dem Sie KI-Agents in der Produktion bereitstellen, skalieren und verwalten können. Er bietet Sicherheitsfunktionen für Unternehmen wie VPC Service Controls und CMEK. Die Infrastruktur wird von Agent Runtime verwaltet, sodass Sie sich auf die Agent-Logik konzentrieren können.

Die Laufzeit für KI-Agenten bietet:

Verwaltete Bereitstellung: Mit einem einzigen SDK-Aufruf können Sie Agenten bereitstellen, die mit dem ADK, LangGraph oder einem beliebigen Python-Framework erstellt wurden.
A2A-Hosting: Agenten als A2A-kompatible Endpunkte mit automatischer Bereitstellung von Agentenkarten und authentifiziertem Zugriff bereitstellen
Persistente Sitzungen: VertexAiSessionService speichert den Unterhaltungsverlauf und den Status über Anfragen hinweg.
Autoscaling: Die Skalierung erfolgt automatisch von null an, um Traffic zu bewältigen. Eine Infrastrukturverwaltung ist nicht erforderlich.
Beobachtbarkeit: Integriertes Tracing, Logging und Monitoring über den Observability-Stack von Google Cloud
Weitere Informationen finden Sie in dieser Dokumentation.

In diesem Codelab stellen Sie den Reservierungsagenten in der Laufzeit für KI-Agenten bereit. Beim Bereitstellungsprozess wird Ihr Agent-Code serialisiert (gepickled) und hochgeladen. Agent Runtime stellt einen serverlosen Endpunkt bereit, der das A2A-Protokoll unterstützt. Andere Agents oder Clients interagieren über standardmäßige HTTP-Aufrufe damit, die mit Google Cloud-Anmeldedaten authentifiziert werden.

5. Reservierungs-Agent erstellen

In diesem Schritt wird ein neuer ADK-Agent erstellt, der Restaurantreservierungen mithilfe des Sitzungsstatus verarbeitet. Der Agent unterstützt drei Vorgänge: „create“, „check“ und „cancel“. Die Telefonnummer dient dabei als Lookup-Schlüssel. Alle Reservierungsdaten befinden sich im Sitzungsstatus des ADK.

Agent erstellen

Verwenden Sie adk create, um die Agent-Verzeichnisstruktur mit der richtigen Modell- und Projektkonfiguration zu generieren:

source .env
uv run adk create reservation_agent \
    --model gemini-2.5-flash \
    --project ${GOOGLE_CLOUD_PROJECT} \
    --region ${GOOGLE_CLOUD_LOCATION}

Dadurch wird ein reservation_agent/-Verzeichnis mit __init__.py, agent.py und .env erstellt, die für das Gemini-Modell auf der Agent Platform vorkonfiguriert sind.

adk-a2a-agent-runtime-starter/
├── reservation_agent/
│   ├── __init__.py
│   ├── agent.py
│   └── .env
├── logs
├── scripts
└── ...

Als Nächstes aktualisieren wir den Agent-Code.

Agentencode schreiben

Öffnen Sie die generierte Agent-Datei:

cloudshell edit reservation_agent/agent.py

Ersetzen Sie dann den Inhalt durch Folgendes:

# 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-Serverkonfiguration vorbereiten

A2A-Agentenkarte definieren

Die Agentenkarte ist eine strukturierte Beschreibung der Funktionen Ihres Agenten. Andere Agenten und Clients verwenden sie, um herauszufinden, was Ihr Agent kann. Kartenkonfiguration erstellen:

cloudshell edit reservation_agent/a2a_config.py

Kopieren Sie Folgendes in 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-Executor erstellen

Der Executor verbindet das A2A-Protokoll und den ADK-Agenten. Er empfängt A2A-Anfragen, führt sie über den ADK-Agenten aus und gibt Ergebnisse als A2A-Aufgaben zurück:

cloudshell edit reservation_agent/executor.py

Kopieren Sie Folgendes in 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())

Der Executor erkennt seine Umgebung automatisch: Wenn GOOGLE_CLOUD_AGENT_ENGINE_ID festgelegt ist (Agent Runtime fügt dies bei der Bereitstellung ein), wird VertexAiSessionService für persistente Sitzungen verwendet. Lokal wird InMemorySessionService als Fallback verwendet.

Das reservation_agent-Verzeichnis sollte nun Folgendes enthalten:

reservation_agent/
├── __init__.py
├── agent.py
├── a2a_config.py
├── executor.py
└── .env

7. A2A-Agent mit dem Agent Platform SDK vorbereiten und lokal testen

In diesem Schritt wird der Reservierungsagent mit der Klasse A2aAgent des Agent Platform SDK ( der SDK-Name verwendet aus Gründen der Abwärtskompatibilität weiterhin den Begriff vertex) als A2A-kompatibler Agent umschlossen. Anschließend wird der vollständige A2A-Protokollablauf lokal getestet – Abrufen der Agentenkarte, Senden von Nachrichten und Abrufen von Aufgaben. Dies ist dasselbe A2aAgent-Objekt, das Sie im nächsten Schritt in der Agent-Laufzeit bereitstellen.

Abhängigkeiten hinzufügen

Installieren Sie das Agent Platform SDK mit Unterstützung für Agent Runtime und ADK sowie das A2A SDK:

uv add "google-cloud-aiplatform[agent_engines,adk]==1.149.0" "a2a-sdk==0.3.26"

A2A-Komponenten

Für das Wrapping eines ADK-Agents für A2A sind drei Komponenten erforderlich:

Agentenkarte: Eine „Visitenkarte“, die die Funktionen, Fähigkeiten und die Endpunkt-URL des Agenten beschreibt. Andere Agenten verwenden diese Informationen, um herauszufinden, was Ihr Agent tut.
Agent Executor: Die Brücke zwischen dem A2A-Protokoll und der Logik Ihres ADK-Agents. Er empfängt A2A-Anfragen, führt sie über den ADK-Agent aus und gibt Ergebnisse als A2A-Aufgaben zurück.
A2aAgent: Die Agent Platform SDK-Klasse, die die Karte und den Executor in einer bereitstellbaren Einheit kombiniert.

Testskript erstellen

Erstellen Sie das folgende Skript, um es lokal zu testen.

cloudshell edit scripts/test_a2a_agent_local.py

Kopieren Sie Folgendes in 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())

Das Testskript importiert die Agent-Karte und den Executor, die Sie im vorherigen Schritt erstellt haben. Es gibt keine Duplikate. Dabei wird ein lokaler A2aAgent erstellt, A2A-Protokollaufrufe werden über Mock-HTTP-Anfragen simuliert und alle drei Reservierungsvorgänge werden überprüft.

Da lokal kein GOOGLE_CLOUD_AGENT_ENGINE_ID festgelegt ist, verwendet der Executor InMemorySessionService. Wenn der Executor in der Agent Runtime bereitgestellt wird, wird automatisch zu VertexAiSessionService für persistente Sitzungen gewechselt.

Test ausführen

PYTHONPATH=. uv run python scripts/test_a2a_agent_local.py

Die Ausgabe umfasst fünf Phasen:

Agentenkarte: Ruft die Funktionen und Skills des Agenten ab.
Reservierung erstellen: Bucht einen Tisch und gibt eine Aufgabe mit der Bestätigung zurück.
Aufgabenergebnis abrufen: Ruft die erledigte Aufgabe mit der Antwort ab.
Reservierung prüfen: Sucht die Reservierung anhand der Telefonnummer
Reservierung stornieren: Die Buchung wird storniert und die Stornierung wird bestätigt.

Beispiel für die Ausgabe (siehe unten)

==================================================
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!
==================================================

An diesem Punkt haben Sie Folgendes überprüft: Die A2A-Agentenkarte beschreibt die richtigen Skills, alle drei Reservierungsvorgänge funktionieren über den Nachrichten-/Aufgabenfluss des A2A-Protokolls und der Status bleibt über Nachrichten im selben Kontext hinweg erhalten.

8. Reservierungs-Agent in der Laufzeit für KI-Agenten bereitstellen

In diesem Schritt wird der Reservierungsagent in der Gemini Enterprise Agent Platform Runtime bereitgestellt. Das ist eine vollständig verwaltete, serverlose Plattform, auf der Ihr Agent gehostet und als sicherer A2A-Endpunkt verfügbar gemacht wird. Nach der Bereitstellung kann jeder autorisierte Client den Agenten über standardmäßige A2A-HTTP-Endpunkte erkennen und mit ihm interagieren.

Staging-Bucket erstellen

Erstellen Sie einen Cloud Storage-Bucket für das Staging der Agent Runtime. Die Agent-Laufzeit verwendet diesen Bucket, um den Code und die Abhängigkeiten Ihres Agenten während der Bereitstellung hochzuladen:

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

Bereitstellungsskript erstellen

Als Nächstes müssen wir das Bereitstellungsskript vorbereiten.

cloudshell edit scripts/deploy_a2a_agent_runtime.py

Kopieren Sie Folgendes in 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()

Das Bereitstellungsskript importiert dieselben agent_card und ReservationAgentExecutor, die auch bei lokalen Tests verwendet werden. Es gibt also keine Codeduplikate. Die Laufzeit für KI-Agenten serialisiert (pickles) das A2aAgent-Objekt zusammen mit seinen Abhängigkeiten für die Bereitstellung. Am Ende des Bereitstellungsskripts wird der Wert RESERVATION_AGENT_RESOURCE_NAME in die Datei .env geschrieben.

In der Agent Runtime bereitstellen

Führen Sie dazu das Bereitstellungsskript aus:

PYTHONPATH=. uv run python scripts/deploy_a2a_agent_runtime.py

Die Bereitstellung dauert 3–5 Minuten. Das Skript stellt einen serverlosen Endpunkt in Agent Runtime bereit, auf dem der Reservierungs-Agent gehostet wird. Nach der erfolgreichen Bereitstellung wird eine Ausgabe wie unten angezeigt.

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

Sie können den bereitgestellten Agent in der Cloud Console ansehen. In der Suchleiste der Console nach Agent Platform suchen

Bewegen Sie dann den Mauszeiger auf dem linken Tab auf Agents und wählen Sie Deployments aus.

Die Reservation Agent wird wie unten dargestellt in der Bereitstellungsliste aufgeführt.

Bereitgestellten KI-Agenten testen

Jetzt können wir den bereitgestellten Agenten testen. Erstellen Sie dazu ein Testskript für den bereitgestellten Agenten:

cloudshell edit scripts/test_a2a_agent_runtime.py

Kopieren Sie Folgendes in 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())

Führen wir den Test durch.

source .env
uv run python scripts/test_a2a_agent_runtime.py

In der Ausgabe wird die Agent-Karte mit dem Skill „Tischreservierungen“ angezeigt, gefolgt von der Erledigung der Aufgabe mit einer Reservierungsbestätigung.

==================================================
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!
==================================================

Der Reservierungs-Agent wird jetzt erfolgreich als verwalteter A2A-Endpunkt in der Agent Runtime ausgeführt.

9. A2A-Reservierungs-Agent in Root-Restaurant-Agent einbinden

In diesem Schritt wird der Restaurant-Agent aktualisiert, damit der bereitgestellte Reservierungs-Agent als Remote-A2A-Sub-Agent verwendet wird. Der Orchestrator wird lokal ausgeführt, während der Reservierungsagent in der Agent Runtime ausgeführt wird. Dies ist eine Teilintegration, mit der die A2A-Verbindung vor der vollständigen Bereitstellung validiert wird.

URL der A2A-Agentenkarte auflösen

RemoteA2aAgent benötigt die Karten-URL des bereitgestellten Reservierungs-KI-Agenten, um seine Funktionen zu ermitteln. Erstellen Sie ein Skript, das diese URL aus der Agent Runtime abruft und in die .env des Restaurant-Agents schreibt:

cloudshell edit scripts/resolve_agent_card_url.py

Kopieren Sie Folgendes in 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())

Führen Sie das Skript aus, um die Datei .env mit der URL der Agent-Karte zu füllen.

uv run python scripts/resolve_agent_card_url.py
source .env

Restaurant-Agent aktualisieren

Öffnen Sie die Datei des Restaurant-Agents:

cloudshell edit restaurant_agent/agent.py

Ersetzen Sie dann den Inhalt durch die aktualisierte Version, die den Remote-Reservierungsagenten als Sub-Agent enthält:

# 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],
)

Die wichtigsten Änderungen gegenüber der vorherigen Version:

GoogleCloudAuth: Ein benutzerdefinierter httpx.Auth-Handler, der das Google Cloud-Zugriffstoken vor jeder Anfrage aktualisiert. Für die Agent-Laufzeit sind authentifizierte A2A-Aufrufe erforderlich und Tokens laufen nach einer bestimmten Zeit ab.
RemoteA2aAgent liest RESERVATION_AGENT_CARD_URL aus dem .env (geschrieben vom Resolve-Script) und verwendet die authentifizierte httpx_client.
Als Sub-Agent registriert – der Orchestrator des ADK delegiert Reservierungsanfragen automatisch an ihn
Anleitung aktualisiert, um die Reservierungsdelegierung zu erwähnen

Integrierten Agenten lokal testen

Für den Starter-Agent ist eine Integration mit der MCP Toolbox erforderlich. Die erforderliche Datei sollte bereits aus einem vorherigen Codelab oder aus dem Starter-Repository bereitgestellt worden sein. Wir müssen nur dafür sorgen, dass der Toolbox-Prozess ordnungsgemäß ausgeführt wird.

Wenn TOOLBOX_URL in Ihrem .env bereits auf einen Cloud Run-Dienst verweist (aus dem vorherigen Codelab oder möglicherweise aus dem full_setup.sh des Starter-Repositorys), können Sie diesen Schritt überspringen. Der Agent stellt dann eine Verbindung zur bereitgestellten Toolbox her.

Wenn Sie stattdessen eine lokale Toolbox benötigen, prüfen Sie, ob bereits eine ausgeführt wird, bevor Sie eine neue Instanz starten:

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

Anschließend können wir versuchen, über die ADK-Web-Entwicklungsoberfläche mit dem Restaurant-Agenten zu interagieren.

uv run adk web --allow_origins "regex:https://.*\.cloudshell\.dev" --port 8080

Öffnen Sie die ADK-Web-UI über die Cloud Shell-Webvorschau (klicken Sie auf die Schaltfläche „Webvorschau“ und ändern Sie den Port zu 8080) und wählen Sie dann restaurant_agent aus.

Gemischte Unterhaltung testen:

Menüanfrage

What Italian dishes do you have?

Reservierungsanfrage

I want to create reservation under name Bob, phone number 123456

Reservierung prüfen

Neue Sitzung erstellen ( neue Unterhaltung beginnen):

Check the reservation for 123456

Beenden Sie den adk web-Prozess mit Strg+C (zweimal). Als Nächstes stellen wir den Agent vollständig bereit, um das System zu vervollständigen.

10. Aktualisierten Restaurant-Agent in Cloud Run bereitstellen

In diesem Schritt wird der Restaurant-Agent mit der A2A-Integration in Cloud Run neu bereitgestellt. Damit ist das vollständig bereitgestellte Multi-Agenten-System abgeschlossen.

Berechtigungen für den Zugriff auf Agent Runtime erteilen

Das Cloud Run-Dienstkonto benötigt die Berechtigung zum Aufrufen der Agent-Laufzeit. Weisen Sie dem Compute Engine-Standarddienstkonto die Rolle roles/aiplatform.user zu:

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"

In Cloud Run bereitstellen

Bei dieser Einrichtung gehen wir davon aus, dass der Restaurant-Agent-Dienst bereits aus dem vorherigen Codelab vorhanden ist oder dass Sie scripts/full_setup.sh ausgeführt haben, wenn Sie von vorn beginnen. Dadurch wird die Bereitstellung mit dem aktualisierten Code (neue RemoteA2aAgent-Integration) wiederholt und die URL der Reservierungsagent-Karte als neue Umgebungsvariable hinzugefügt. Vorhandene Umgebungsvariablen (TOOLBOX_URL, GOOGLE_CLOUD_PROJECT usw.) bleiben erhalten:

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

Vollständig bereitgestelltes System testen

Rufen Sie die URL des bereitgestellten Dienstes ab:

AGENT_URL=$(gcloud run services describe restaurant-agent --region=$REGION --format='value(status.url)')
echo "Agent URL: $AGENT_URL"

Rufen Sie die URL in Ihrem Browser auf. Die ADK-Web-UI wird geladen. Das ist dieselbe Benutzeroberfläche, die Sie lokal verwendet haben, jetzt aber in Cloud Run ausgeführt wird.

Sie können sich gern mit dem Kundenservicemitarbeiter unterhalten.

Menüanfrage

What spicy dishes do you have?

Reservierungsanfrage

Book a table for 4 on Friday at 7pm. Name: Eve, Phone: 555-0505

Reservierung prüfen

Neue Sitzung erstellen ( neue Unterhaltung beginnen):

Check reservation for 555-0505

Das Multi-Agenten-System ist vollständig bereitgestellt. Der Restaurant-Agent in Cloud Run koordiniert zwei Backend-Dienste: die MCP-Toolbox für Menüvorgänge und den A2A-Reservierungs-Agent in der Agent Runtime.

11. Glückwunsch!

Sie haben ein Multi-Agenten-System mit dem A2A-Protokoll in Google Cloud erstellt und bereitgestellt.

Lerninhalte

Sie haben einen ADK-Agenten erstellt, der den Sitzungsstatus (ToolContext) verwendet, um Reservierungsdaten ohne Datenbank zu verwalten.
Sie haben einen A2A-Agenten mit dem Agent Platform SDK in der Agent-Laufzeit bereitgestellt.
Ein Remote-A2A-Agent wurde von einem anderen ADK-Agenten mit RemoteA2aAgent als Sub-Agenten verwendet.
Das System wurde inkrementell getestet: lokales A2A → bereitgestelltes A2A → Teilintegration → vollständige Bereitstellung.

Aufräumen

Löschen Sie die in diesem Codelab erstellten Ressourcen, um zu vermeiden, dass Ihrem Google Cloud-Konto Gebühren in Rechnung gestellt werden.

Option 1: Projekt löschen (empfohlen)

gcloud projects delete $GOOGLE_CLOUD_PROJECT

Option 2: Einzelne Ressourcen löschen

# 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