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

Agenci na dużą skalę: architektura wieloagentowa z protokołem A2A w środowisku wykonawczym agentów i integracja z pakietem ADK

1. Wprowadzenie

W miarę jak agenty AI przejmują coraz więcej obowiązków, utrzymanie, skalowanie i rozwijanie jednego agenta, który robi wszystko, staje się trudne. Różne funkcje często wymagają różnych strategii wdrażania, cykli aktualizacji, a nawet różnych zespołów odpowiedzialnych za nie.

Protokół A2A (Agent2Agent) rozwiązuje problem komunikacji – standaryzuje sposób, w jaki agenty odkrywają swoje możliwości i współpracują w ramach różnych platform i organizacji.
Gemini Enterprise Agent Platform Runtime rozwiązuje problem z wdrażaniem. Jest to w pełni zarządzana platforma bezserwerowa, która hostuje Twoich agentów z wbudowaną obsługą A2A, autoskalowaniem, bezpiecznymi punktami końcowymi, trwałymi sesjami i zerową obsługą infrastruktury.

Umożliwiają one tworzenie wyspecjalizowanych agentów, wdrażanie ich jako wykrywalnych usług A2A i łączenie ich w systemy wieloagentowe.

Co utworzysz

Agent rezerwacji, który zarządza rezerwacjami stolików w restauracji (tworzy, sprawdza i anuluje rezerwacje) za pomocą stanu sesji pakietu ADK zarządzanego przez sesje Gemini Enterprise Agent Platform. Wdrażasz tego agenta w środowisku wykonawczym Gemini Enterprise Agent Platform, gdzie staje się on widoczny na karcie agenta protokołu A2A. Następnie ulepszysz agenta concierge restauracji Foodie Finds (z wymaganego ćwiczenia, nie martw się, jeśli nie znasz tego ćwiczenia – przygotowaliśmy dla Ciebie repozytorium startowe), aby korzystał z agenta rezerwacji jako zdalnego sub-agenta A2A. W rezultacie powstaje system wieloagentowy, w którym koordynator kieruje zapytania dotyczące menu do zestawu narzędzi MCP, a prośby o rezerwację do zdalnego agenta A2A.

Czego się nauczysz

Tworzenie agenta ADK, który korzysta z usługi zarządzania sesjami do zarządzania danymi rezerwacji
Udostępnianie agenta ADK jako serwera A2A z kartami i umiejętnościami agenta
Wdrażanie agenta A2A w środowisku wykonawczym agentów Gemini Enterprise
Korzystanie ze zdalnego agenta A2A z innego agenta ADK za pomocą RemoteA2aAgent i obsługa uwierzytelnionego żądania
Testowanie systemów wieloagentowych przyrostowo: lokalny protokół A2A, wdrożony protokół A2A, częściowa integracja, pełne wdrożenie

Wymagania wstępne

(Zalecane) Ukończone te codelaby :
Tworzenie trwałych agentów AI z użyciem pakietu ADK i CloudSQL –> więcej informacji o sesji i stanie pakietu ADK
Agentowy RAG z ADK, MCP Toolbox i Cloud SQL – możesz kontynuować tworzenie agenta na podstawie tego ćwiczenia. Podany kod startowy jest identyczny.
konto Google Cloud z aktywnym kontem rozliczeniowym;
podstawowa znajomość Pythona i koncepcji ADK,

2. Konfiguracja środowiska – kontynuacja poprzednich ćwiczeń z programowania

Opisy, które znajdziesz w tym ćwiczeniu, są kontynuacją tego ćwiczenia wstępnego: Agentowy RAG z ADK, MCP Toolbox i Cloud SQL . Możesz kontynuować pracę z poprzednich warsztatów

Możemy zacząć tworzyć w poprzednim katalogu roboczym z ćwiczeń ( katalog roboczy powinien mieć nazwę build-agent-adk-toolbox-cloudsql). Aby uniknąć nieporozumień, zmieńmy nazwę katalogu na taką samą, jakiej używamy, gdy zaczynamy od nowa.

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

Sprawdź, czy kluczowe pliki z poprzednich ćwiczeń z programowania są na swoim miejscu:

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

Powinien pojawić się plik restaurant_agent/agent.py z importem LlmAgent oraz plik tools.yaml z konfiguracją Zestawu narzędzi.

Następnie ponownie zainicjujmy środowisko Pythona.

rm -rf .venv
uv sync

Sprawdź też, czy baza danych jest wypełniona i gotowa:

uv run python scripts/verify_seed.py

Jeśli będziesz postępować zgodnie ze szczegółami testowania z poprzedniego laboratorium, możesz zobaczyć takie dane wyjściowe:

Menu Items: 16/15
Embeddings: 16/15

✗ Database not ready

W porządku. Sprawdzanie bazy danych nie uwzględnia dodatkowych danych wprowadzonych podczas sprawdzania pozyskiwania danych. Jeśli masz co najmniej 15 danych, wszystko jest w porządku.

Aktywowanie wymaganego interfejsu API

Następnie musimy włączyć wymagany interfejs API, aby korzystać z Gemini Enterprise Agent Platform.

gcloud services enable \
  cloudresourcemanager.googleapis.com

Powinieneś(-aś) mieć już niezbędne pliki i infrastrukturę, aby przejść do następnej sekcji: A2A Protocol and Gemini Enterprise Agent Runtime!

3. Konfiguracja środowiska – świeży start z repozytorium początkowym

Ten krok przygotowuje środowisko Cloud Shell, konfiguruje projekt Google Cloud i klonuje repozytorium początkowe.

Otwieranie Cloud Shell

Otwórz Cloud Shell w przeglądarce. Cloud Shell zapewnia wstępnie skonfigurowane środowisko ze wszystkimi narzędziami potrzebnymi do tego ćwiczenia. Gdy pojawi się prośba o autoryzację, kliknij Autoryzuj.

Następnie kliknij „Widok” –> „Terminal”, aby otworzyć terminal.Interfejs powinien wyglądać podobnie do tego:

Będzie to nasz główny interfejs: IDE u góry, terminal u dołu.

Konfigurowanie katalogu roboczego

Sklonuj repozytorium początkowe. Będzie ono zawierać cały kod, który napiszesz w tym ćwiczeniu:

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

Utwórz plik .env na podstawie podanego szablonu:

cp .env.example .env

Aby uprościć konfigurację projektu w terminalu, pobierz ten skrypt konfiguracji projektu do katalogu roboczego:

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

Uruchom skrypt. Weryfikuje ono Twoje próbne konto rozliczeniowe, tworzy nowy projekt (lub weryfikuje istniejący), zapisuje identyfikator projektu w pliku .env w bieżącym katalogu i ustawia aktywny projekt w gcloud.

bash setup_verify_trial_project.sh && source .env

Skrypt:

Sprawdź, czy masz aktywne konto rozliczeniowe w wersji próbnej
Sprawdź, czy w .env istnieje projekt (jeśli tak)
Utwórz nowy projekt lub użyj istniejącego.
Połącz próbne konto rozliczeniowe z projektem
Zapisz identyfikator projektu w .env.
Ustaw projekt jako aktywny projekt gcloud.

Sprawdź, czy projekt jest prawidłowo ustawiony, sprawdzając żółty tekst obok katalogu roboczego w wierszu poleceń terminala Cloud Shell. Powinien wyświetlać identyfikator projektu.

Aktywowanie wymaganego interfejsu API

Następnie musimy włączyć wymagany interfejs API, aby korzystać z Gemini Enterprise Agent Platform.

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

Konfiguracja infrastruktury dla początkujących

Najpierw musimy zainstalować zależności Pythona za pomocą uv, czyli szybkiego menedżera pakietów i projektów Pythona napisanego w języku Rust ( dokumentacja uv). W tym ćwiczeniu używamy go ze względu na szybkość i prostotę utrzymywania projektu w Pythonie.

uv sync

Następnie uruchom pełny skrypt konfiguracji, który utworzy instancję Cloud SQL, wypełni ją danymi i wdroży usługę Toolbox, która będzie działać jako stan początkowy agenta restauracji.

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

4. Koncepcja: protokół Agent2Agent (A2A) i środowisko wykonawcze agentów Gemini Enterprise

Zanim zaczniemy, poświęćmy chwilę na zapoznanie się z 2 kluczowymi technologiami, które są przedstawione w tym laboratorium, aby skalować aplikację opartą na agentach.

Protokół Agent2Agent (A2A)

Protokół Agent2Agent (A2A) to otwarty standard zaprojektowany z myślą o umożliwieniu płynnej komunikacji i współpracy między agentami AI. MCP (Model Context Protocol) łączy agenty z narzędziami i danymi, a A2A łączy agenty z innymi agentami, umożliwiając im odkrywanie swoich możliwości, delegowanie zadań i współpracę w ramach różnych platform i organizacji.

Główna różnica między opakowaniem agenta jako narzędzia (za pomocą MCP) a udostępnianiem go za pomocą A2A polega na tym, że narzędzia są bezstanowe i wykonują pojedyncze funkcje, a agenci A2A mogą wnioskować, utrzymywać stan i obsługiwać interakcje wieloetapowe, takie jak negocjacje czy wyjaśnienia. Agent udostępniony za pomocą A2A zachowuje pełną funkcjonalność, zamiast być ograniczony do wywołania funkcji.

A2A definiuje 3 podstawowe pojęcia:

Karta agenta – dokument JSON opisujący, co robi agent, jakie ma umiejętności i jaki jest jego punkt końcowy. Inni agenci pobierają tę kartę, aby poznać możliwości.
Wiadomość – prośba użytkownika lub agenta wysłana do punktu końcowego A2A, która uruchamia zadanie.
Zadanie – jednostka pracy z określonym cyklem życia (przesłane → w trakcie realizacji → ukończone/nieudane) i artefaktami zawierającymi wyniki.

Więcej informacji znajdziesz w artykule Co to jest A2A?

Środowisko wykonawcze Gemini Enterprise Agent Platform

Agent Runtime to w pełni zarządzana usługa w Google Cloud, która umożliwia wdrażanie, skalowanie i zarządzanie agentami AI w środowisku produkcyjnym z funkcjami zabezpieczeń klasy korporacyjnej (np. ustawieniami usługi VPC i kluczami szyfrowania zarządzanymi przez klienta). Obsługuje infrastrukturę, dzięki czemu możesz skupić się na logice agenta.

Środowisko wykonawcze agentów zapewnia:

Zarządzane wdrażanie – wdrażaj agentów utworzonych za pomocą pakietu ADK, LangGraph lub dowolnej platformy Python za pomocą jednego wywołania pakietu SDK.
Hosting A2A – wdrażanie agentów jako punktów końcowych zgodnych z A2A z automatycznym wyświetlaniem karty agenta i uwierzytelnionym dostępem.
Sesje trwałe – VertexAiSessionService przechowują historię rozmowy i stan w różnych żądaniach.
Autoskalowanie – skalowanie od zera w celu obsługi ruchu bez zarządzania infrastrukturą.
Dostrzegalność – wbudowane śledzenie, logowanie i monitorowanie za pomocą pakietu narzędzi do dostrzegalności Google Cloud.
i wiele innych funkcji. Więcej informacji znajdziesz w tej dokumentacji.

W tym ćwiczeniu wdrożysz agenta rezerwacji w środowisku wykonawczym agentów. Proces wdrażania serializuje (pickles) kod agenta i przesyła go. Środowisko wykonawcze agenta udostępnia bezserwerowy punkt końcowy, który obsługuje protokół A2A. Inne agenty (lub klienci) wchodzą z nim w interakcję za pomocą standardowych wywołań HTTP uwierzytelnianych za pomocą danych logowania Google Cloud.

5. Tworzenie agenta rezerwacji

W tym kroku utworzysz nowego agenta ADK, który obsługuje rezerwacje w restauracjach za pomocą stanu sesji. Agent obsługuje 3 operacje: tworzenie, sprawdzanie i anulowanie. Numer telefonu jest kluczem wyszukiwania. Wszystkie dane rezerwacji znajdują się w stanie sesji ADK.

Utwórz szkielet agenta

Użyj polecenia adk create, aby wygenerować strukturę katalogu agenta z prawidłową konfiguracją modelu i projektu:

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

Spowoduje to utworzenie katalogu reservation_agent/ ze wstępnie skonfigurowanymi plikami __init__.py, agent.py i .env dla modelu Gemini na platformie Agent Platform.

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

Następnie zaktualizujmy kod agenta

Napisz kod agenta

Otwórz wygenerowany plik agenta:

cloudshell edit reservation_agent/agent.py

Następnie zastąp zawartość tym kodem:

# 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. Przygotowywanie konfiguracji serwera A2A

Zdefiniuj kartę agenta A2A

Karta agenta to strukturalny opis możliwości agenta. Inni agenci i klienci używają jej, aby dowiedzieć się, do czego służy Twój agent. Utwórz konfigurację karty:

cloudshell edit reservation_agent/a2a_config.py

Skopiuj poniższy tekst do pliku 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],
)

Utwórz wykonawcę A2A

Wykonawca łączy protokół A2A z agentem pakietu ADK. Otrzymuje żądania A2A, przetwarza je za pomocą agenta ADK i zwraca wyniki jako zadania A2A:

cloudshell edit reservation_agent/executor.py

Skopiuj poniższy tekst do pliku 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())

Wykonawca automatycznie wykrywa swoje środowisko: gdy ustawiona jest wartość GOOGLE_CLOUD_AGENT_ENGINE_ID (środowisko wykonawcze agenta wstrzykuje ją w momencie wdrażania), używa VertexAiSessionService w przypadku trwałych sesji. Lokalnie wraca do InMemorySessionService.

Katalog reservation_agent powinien teraz zawierać:

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

7. Przygotowywanie agenta A2A za pomocą pakietu SDK usługi Agent Platform i testowanie go lokalnie

W tym kroku agent rezerwacji jest opakowywany jako agent zgodny z A2A przy użyciu klasy A2aAgent pakietu SDK Agent Platform ( nazwa pakietu SDK nadal używa terminu vertex ze względu na zgodność wsteczną), a następnie lokalnie testowany jest pełny przepływ protokołu A2A – pobieranie karty agenta, wysyłanie wiadomości i pobieranie zadań. Jest to ten sam obiekt A2aAgent, który wdrożysz w środowisku wykonawczym agenta w następnym kroku.

Dodawanie zależności

Zainstaluj pakiet SDK usługi Agent Platform z obsługą środowiska wykonawczego agenta i pakietu ADK oraz pakiet SDK A2A:

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

Omówienie komponentów A2A

Owijanie agenta ADK na potrzeby A2A wymaga 3 komponentów:

Karta agenta – „wizytówka”, która opisuje możliwości, umiejętności i adres URL agenta. Inni agenci używają jej, aby dowiedzieć się, co robi Twój agent.
Agent Executor – pomost między protokołem A2A a logiką agenta ADK. Otrzymuje żądania A2A, przetwarza je za pomocą agenta ADK i zwraca wyniki jako zadania A2A.
A2aAgent – klasa pakietu SDK usługi Agent Platform, która łączy kartę i wykonawcę w jednostkę do wdrożenia.

Tworzenie scenariusza testowania

Utwórz ten skrypt, aby przetestować go lokalnie:

cloudshell edit scripts/test_a2a_agent_local.py

Skopiuj poniższy tekst do pliku 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())

Scenariusz testowania importuje kartę agenta i wykonawcę utworzone w poprzednim kroku – nie ma duplikatów. Utworzy lokalny plik A2aAgent, zasymuluje wywołania protokołu A2A za pomocą przykładowych żądań HTTP i zweryfikuje wszystkie 3 operacje rezerwacji.

Ponieważ lokalnie nie ustawiono GOOGLE_CLOUD_AGENT_ENGINE_ID, wykonawca używa InMemorySessionService. Po wdrożeniu w środowisku wykonawczym agenta ten sam wykonawca automatycznie przełącza się na VertexAiSessionService w przypadku trwałych sesji.

Przeprowadź test

PYTHONPATH=. uv run python scripts/test_a2a_agent_local.py

Dane wyjściowe obejmują 5 etapów:

Karta agenta – pobiera możliwości i umiejętności agenta.
Utwórz rezerwację – rezerwuje stolik i zwraca zadanie z potwierdzeniem.
Pobierz wynik zadania – pobiera ukończone zadanie z odpowiedzią.
Sprawdź rezerwację – wyszukuje rezerwację według numeru telefonu.
Anuluj rezerwację – anuluje rezerwację i potwierdza anulowanie.

Przykład danych wyjściowych, jak pokazano poniżej

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

Na tym etapie masz już pewność, że karta agenta A2A zawiera prawidłowe umiejętności, wszystkie 3 operacje rezerwacji działają w ramach przepływu wiadomości/zadań protokołu A2A, a stan jest zachowywany w wiadomościach w tym samym kontekście.

8. Wdróż agenta rezerwacji w środowisku wykonawczym agentów

W tym kroku wdrażasz agenta rezerwacji w środowisku wykonawczym platformy agentów Gemini Enterprise – w pełni zarządzanej platformie bezserwerowej, która hostuje agenta i udostępnia go jako bezpieczny punkt końcowy A2A. Po wdrożeniu każdy autoryzowany klient może odkrywać agenta i wchodzić z nim w interakcje za pomocą standardowych punktów końcowych HTTP A2A.

Utwórz zasobnik tymczasowy

Utwórz zasobnik Cloud Storage na potrzeby przygotowania środowiska wykonawczego agentów. Środowisko wykonawcze agentów używa tego zasobnika do przesyłania kodu i zależności agenta podczas wdrażania:

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

Tworzenie skryptu wdrożenia

Następnie musimy przygotować skrypt wdrażania.

cloudshell edit scripts/deploy_a2a_agent_runtime.py

Skopiuj poniższy tekst do pliku 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()

Skrypt wdrożenia importuje te same obiekty agent_card i ReservationAgentExecutor, które są używane w testach lokalnych – nie ma duplikowania kodu. Środowisko wykonawcze agenta serializuje (pickles) obiekt A2aAgent wraz z jego zależnościami na potrzeby wdrożenia. Na końcu skryptu wdrożenia zapisuje wartość RESERVATION_AGENT_RESOURCE_NAME w pliku .env.

Wdrażanie w środowisku wykonawczym agentów

Uruchom skrypt wdrażania:

PYTHONPATH=. uv run python scripts/deploy_a2a_agent_runtime.py

Wdrożenie zajmuje 3–5 minut. Skrypt udostępnia bezserwerowy punkt końcowy w środowisku wykonawczym agentów, który hostuje agenta rezerwacji. Po udanym wdrożeniu zobaczysz dane wyjściowe podobne do tych poniżej.

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

Wdrożonego agenta możesz wyświetlić w konsoli w chmurze. Na pasku wyszukiwania w konsoli wpisz Agent Platform.

Następnie na karcie po lewej stronie najedź kursorem na Agents i wybierz Deployments.

Na liście wdrożeń zobaczysz Reservation Agent, jak pokazano poniżej.

Testowanie wdrożonego agenta

Teraz możemy przetestować wdrożonego agenta. W tym celu utwórz scenariusz testowania:

cloudshell edit scripts/test_a2a_agent_runtime.py

Skopiuj poniższy tekst do pliku 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())

Następnie przeprowadźmy test.

source .env
uv run python scripts/test_a2a_agent_runtime.py

Wynik pokazuje kartę agenta z umiejętnością „Rezerwacje w restauracjach”, a następnie wykonanie zadania z potwierdzeniem rezerwacji.

==================================================
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 rezerwacji działa teraz jako zarządzany punkt końcowy A2A w środowisku wykonawczym agenta.

9. Integrowanie agenta rezerwacji A2A z agentem głównym restauracji

Ten krok powoduje uaktualnienie agenta restauracji, aby używał wdrożonego agenta rezerwacji jako zdalnego subagenta A2A. Koordynator działa lokalnie, a agent rezerwacji działa w środowisku wykonawczym agenta – jest to częściowa integracja, która weryfikuje połączenie A2A przed pełnym wdrożeniem.

Rozwiązywanie problemów z adresem URL karty agenta A2A

RemoteA2aAgent potrzebuje adresu URL karty wdrożonego agenta rezerwacji, aby wykryć jego możliwości. Utwórz skrypt, który pobiera ten adres URL ze środowiska wykonawczego agenta i zapisuje go w .env agenta restauracji:

cloudshell edit scripts/resolve_agent_card_url.py

Skopiuj poniższy tekst do pliku 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())

Uruchom skrypt, aby wypełnić plik .env adresem URL karty agenta.

uv run python scripts/resolve_agent_card_url.py
source .env

Aktualizowanie agenta restauracji

Otwórz plik agenta restauracji:

cloudshell edit restaurant_agent/agent.py

Następnie zastąp zawartość zaktualizowaną wersją, która zawiera zdalnego agenta rezerwacji jako sub-agenta:

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

Najważniejsze zmiany w porównaniu z poprzednią wersją:

GoogleCloudAuth – niestandardowy moduł obsługi httpx.Auth, który odświeża token dostępu Google Cloud przed każdym żądaniem. Środowisko wykonawcze agenta wymaga uwierzytelnionych wywołań A2A, a tokeny wygasają po pewnym czasie.
RemoteA2aAgent odczytuje RESERVATION_AGENT_CARD_URL z .env (zapisanego przez skrypt resolve) i używa uwierzytelnionego httpx_client.
Zarejestrowany jako subagent – aranżer ADK automatycznie przekazuje do niego prośby o rezerwację.
Zaktualizowane instrukcje dotyczące przekazywania rezerwacji

Testowanie zintegrowanego agenta lokalnie

Agent początkowy wymaga integracji z MCP Toolbox. Wymagany plik powinien być już dostępny z poprzednich ćwiczeń lub z repozytorium początkowego. Musimy tylko upewnić się, że proces narzędziowy działa prawidłowo.

Jeśli TOOLBOX_URL w .env wskazuje już usługę Cloud Run (z poprzedniego laboratorium lub z full_setup.sh w repozytorium początkowym), możesz pominąć ten krok – agent połączy się z wdrożonym zestawem narzędzi.

Jeśli potrzebujesz lokalnego pakietu Toolbox, przed uruchomieniem nowej instancji sprawdź, czy jest już uruchomiona:

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

Następnie możemy spróbować wejść w interakcję z agentem restauracji za pomocą interfejsu internetowego ADK.

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

Otwórz interfejs ADK w przeglądarce za pomocą podglądu w przeglądarce Cloud Shell (kliknij przycisk Podgląd w przeglądarce i zmień port na 8080), a następnie wybierz restaurant_agent.

Przetestuj rozmowę mieszaną:

Zapytanie do menu

What Italian dishes do you have?

Prośba o rezerwację

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

Sprawdź rezerwację

Utwórz nową sesję ( rozpocznij nową rozmowę):

Check the reservation for 123456

Zatrzymaj proces adk web, naciskając dwukrotnie Ctrl+C. Następnie dokończmy system, w pełni wdrażając agenta

10. Wdrażanie zaktualizowanego agenta restauracji w Cloud Run

Ten krok ponownie wdraża agenta restauracji w Cloud Run z integracją A2A, co powoduje pełne wdrożenie systemu wieloagentowego.

Przyznawanie uprawnień dostępu do środowiska wykonawczego agenta

Konto usługi Cloud Run musi mieć uprawnienia do wywoływania środowiska wykonawczego agenta. Przyznaj rolę roles/aiplatform.user domyślnemu kontu usługi Compute Engine:

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"

Wdróż w Cloud Run

W tej konfiguracji zakładamy, że usługa agenta restauracji już istnieje z poprzedniego laboratorium lub została utworzona przez uruchomienie polecenia scripts/full_setup.sh, jeśli zaczynasz od nowa. Spowoduje to ponowne wdrożenie z zaktualizowanym kodem (nowa integracja RemoteA2aAgent) i dodanie adresu URL karty agenta rezerwacji jako nowej zmiennej środowiskowej. Istniejące zmienne środowiskowe (TOOLBOX_URL, GOOGLE_CLOUD_PROJECT itp.) zostaną zachowane:

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

Testowanie w pełni wdrożonego systemu

Pobierz adres URL wdrożonej usługi:

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

Otwórz adres URL w przeglądarce. Załaduje się interfejs internetowy ADK – to ten sam interfejs, którego używasz lokalnie, ale teraz działa w Cloud Run.

Możesz swobodnie rozmawiać z pracownikiem obsługi klienta.

Zapytanie do menu

What spicy dishes do you have?

Prośba o rezerwację

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

Sprawdź rezerwację

Utwórz nową sesję ( rozpocznij nową rozmowę):

Check reservation for 555-0505

System wieloagentowy jest w pełni wdrożony. Agent restauracji w Cloud Run koordynuje działania między 2 usługami backendu: MCP Toolbox do obsługi menu i agentem rezerwacji A2A w Agent Runtime.

11. Gratulacje!

Udało Ci się utworzyć i wdrożyć system wieloagentowy przy użyciu protokołu A2A w Google Cloud.

Czego się nauczysz

Tworzenie agenta ADK, który używa stanu sesji (ToolContext) do zarządzania danymi rezerwacji bez bazy danych
wdrożono agenta A2A w środowisku wykonawczym agenta za pomocą pakietu SDK usługi Agent Platform;
Wykorzystanie zdalnego agenta A2A z innego agenta ADK za pomocą RemoteA2aAgent jako sub-agenta
Stopniowe testowanie systemu: lokalny protokół A2A → wdrożony protokół A2A → częściowa integracja → pełne wdrożenie

Czyszczenie danych

Aby uniknąć obciążenia konta Google Cloud opłatami, usuń zasoby utworzone w tym laboratorium.

Opcja 1. Usuwanie projektu (zalecane)

gcloud projects delete $GOOGLE_CLOUD_PROJECT

Opcja 2. Usuwanie poszczególnych zasobów

# 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