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 wykonuje wszystkie zadania, 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.
Środowisko wykonawcze platformy agentów Gemini Enterprise rozwiązuje problem wdrażania – 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 zerowym zarządzaniem infrastrukturą.

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 (tworzenie, sprawdzanie i anulowanie) za pomocą stanu sesji pakietu ADK zarządzanego przez sesje na platformie agentów Gemini Enterprise. Wdrażasz tego agenta w środowisku wykonawczym platformy agentów Gemini Enterprise, gdzie staje się on widoczny na karcie agenta protokołu A2A. Następnie ulepszysz agenta concierge restauracji Foodie Finds (z wymaganego ćwiczenia w Codelabs, nie martw się, jeśli jeszcze go nie znasz – 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 o menu do MCP Toolbox, 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
Stopniowe testowanie systemów wieloagentowych: 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 poprzedniego ćwiczenia

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 zainicjuj ś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 platformy agentów Gemini Enterprise.

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 wykona te działania:

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 platformy agentów Gemini Enterprise.

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 pełnić rolę początkowego stanu 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ć naszą 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 różnych ramach i organizacjach.

Kluczowa różnica między opakowaniem agenta jako narzędzia (za pomocą MCP) a udostępnianiem go za pomocą A2A: 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 działanie agenta, jego umiejętności i punkt końcowy. Inne agenty pobierają tę kartę, aby odkrywać 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 → ukończone/nieudane) i artefaktami zawierającymi wyniki.

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

Środowisko wykonawcze platformy agentów Gemini Enterprise

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. Ustawienia usługi VPC i CMEK). 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 protokołem A2A z automatycznym udostępnianiem kart agenta i uwierzytelnionym dostępem.
Sesje trwałe – VertexAiSessionService przechowują historię rozmów i stan w różnych żądaniach.
Autoskalowanie – skaluje się od zera, aby obsługiwać ruch, 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

Ten krok tworzy 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 są przechowywane 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-2.5-flash \
    --project ${GOOGLE_CLOUD_PROJECT} \
    --region ${GOOGLE_CLOUD_LOCATION}

Spowoduje to utworzenie katalogu reservation_agent/ z plikami __init__.py, agent.py i .env wstępnie skonfigurowanymi 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
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. 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 do pliku reservation_agent/a2a_config.py ten tekst:

# 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, przekazuje je przez agenta ADK i zwraca wyniki jako zadania A2A:

cloudshell edit reservation_agent/executor.py

Skopiuj do pliku reservation_agent/executor.py ten tekst:

# 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 sesji trwałych. Lokalnie jest to 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 platformy agentów ( 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 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” opisująca możliwości, umiejętności i adres URL punktu końcowego agenta. Inni agenci używają tego opisu, aby dowiedzieć się, do czego służy 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 przeprowadzić test lokalny:

cloudshell edit scripts/test_a2a_agent_local.py

Skopiuj do pliku scripts/test_a2a_agent_local.py ten tekst:

# 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 — bez duplikowania. 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

Ten krok wdraża 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 wykrywać 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 agenta. Ś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 do pliku scripts/deploy_a2a_agent_runtime.py ten tekst:

# 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 wdrażania importuje te same pliki agent_card i ReservationAgentExecutor, które są używane w testach lokalnych, więc nie ma duplikacji kodu. Środowisko wykonawcze agentów serializuje (pickles) obiekt A2aAgent wraz z jego zależnościami na potrzeby wdrożenia. Na końcu skryptu wdrażania zapisze on 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, w którym jest hostowany agent 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 Google Cloud. 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 do pliku scripts/test_a2a_agent_runtime.py ten tekst:

# 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ź 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 prawidłowo jako zarządzany punkt końcowy A2A w środowisku wykonawczym agenta.

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

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

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

RemoteA2aAgent potrzebuje adresu URL karty wdrożonego agenta rezerwacji, aby poznać 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 do pliku scripts/resolve_agent_card_url.py ten tekst:

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

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 zajęć lub z repozytorium początkowego. Musimy tylko upewnić się, że proces w zestawie narzędzi przebiega 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 narzędziem.

Jeśli potrzebujesz lokalnego pakietu narzędzi, przed uruchomieniem nowej instancji sprawdź, czy jest już uruchomiona inna:

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 Cloud Shell (kliknij przycisk Podgląd w przeglądarce i zmień port na 8080), a następnie wybierz restaurant_agent.

Przetestuj rozmowę mieszaną:

Zapytanie o 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, wdrażając w pełni 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 o 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łanie 2 usług backendu: MCP Toolbox do obsługi menu i agenta rezerwacji A2A w środowisku wykonawczym agenta.

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 przy użyciu RemoteA2aAgent jako sub-agenta
System był testowany stopniowo: 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