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

Agentes a gran escala: Arquitectura multiagente con el protocolo A2A en el tiempo de ejecución de Agent y la integración del ADK

1. Introducción

A medida que los agentes de IA asumen más responsabilidades, se vuelve difícil mantener, escalar y evolucionar un solo agente que haga todo. Las diferentes capacidades suelen requerir distintas estrategias de implementación, ciclos de actualización o incluso diferentes equipos que las administren.

El protocolo A2A (Agent2Agent) resuelve el problema de la comunicación, ya que estandariza la forma en que los agentes descubren las capacidades de los demás y colaboran en diferentes frameworks y organizaciones.
Gemini Enterprise Agent Platform Runtime resuelve el problema de la implementación, ya que es una plataforma sin servidores completamente administrada que aloja tus agentes con compatibilidad integrada de A2A, ajuste de escala automático, extremos seguros, sesiones persistentes y administración de infraestructura nula.

Juntos, te permiten crear agentes especializados, implementarlos como servicios A2A detectables y componerlos en sistemas multiagente.

Qué compilarás

Un agente de reservas que administra las reservas de mesas de restaurantes (crea, verifica y cancela) con el estado de la sesión del ADK que administra Gemini Enterprise Agent Platform Sessions. Implementas este agente en Gemini Enterprise Agent Platform Runtime, donde se puede descubrir a través de la tarjeta de agente del protocolo A2A. Luego, actualizarás el agente de asistente de restaurante Foodie Finds (del codelab de requisitos previos; no te preocupes si no lo visitaste, preparamos un repositorio de inicio para ti) para que consuma el agente de reservas como un agente secundario de A2A remoto. El resultado es un sistema multiagente en el que el orquestador enruta las consultas del menú a la caja de herramientas de MCP y las solicitudes de reserva al agente A2A remoto.

Qué aprenderás

Compila un agente del ADK que use el servicio de sesión administrada para administrar los datos de reserva
Cómo exponer un agente del ADK como servidor A2A con tarjetas y habilidades de agente
Implementa un agente de A2A en el entorno de ejecución de agentes de Gemini Enterprise
Consume un agente A2A remoto desde otro agente del ADK con RemoteA2aAgent y controla la solicitud autenticada
Prueba los sistemas multiagente de forma incremental: A2A local, A2A implementado, integración parcial, implementación completa

Requisitos previos

(Recomendado) Completaste los siguientes codelabs :
Building Persistent AI Agents with ADK and CloudSQL: Más detalles sobre la sesión y el estado del ADK
RAG agente con ADK, MCP Toolbox y Cloud SQL: Puedes seguir compilando tu agente a partir de este codelab, ya que el código de partida proporcionado es idéntico.
Una cuenta de Google Cloud con una cuenta de facturación activa
Conocimientos básicos sobre Python y los conceptos del ADK

2. Configuración del entorno: Continuación del codelab anterior

Las narrativas que proporcionamos en este codelab son, en realidad, la continuación de este codelab de requisitos previos: RAG agente con ADK, MCP Toolbox y Cloud SQL . Puedes continuar tu trabajo desde el codelab anterior.

Podemos comenzar a compilar en el directorio de trabajo del codelab anterior ( el directorio de trabajo debería ser build-agent-adk-toolbox-cloudsql). Para evitar confusiones, cambiemos el nombre del directorio por el mismo nombre que usamos cuando comenzamos de cero.

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

Verifica que los archivos de claves del codelab anterior estén en su lugar:

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

Deberías ver el archivo restaurant_agent/agent.py con la importación de LlmAgent y el archivo tools.yaml con la configuración de tu Toolbox.

A continuación, reinicializaremos nuestro entorno de Python.

rm -rf .venv
uv sync

Además, verifica que la base de datos se haya inicializado y esté lista:

uv run python scripts/verify_seed.py

Si sigues todos los detalles de las pruebas del codelab anterior, es posible que veas un resultado como este:

Menu Items: 16/15
Embeddings: 16/15

✗ Database not ready

¡No hay problema! La verificación de la base de datos no tiene en cuenta los datos adicionales que ingresas desde la verificación de la transferencia de datos. Siempre y cuando tengas más de 15 datos, todo estará bien.

Activa la API requerida

A continuación, deberemos asegurarnos de habilitar la API necesaria para interactuar con Agent Platform de Gemini Enterprise

gcloud services enable \
  cloudresourcemanager.googleapis.com

Ya deberías tener los archivos y la infraestructura necesarios para continuar con la siguiente sección: A2A Protocol and Gemini Enterprise Agent Runtime.

3. Configuración del entorno: Comienza de cero con el repo de inicio

En este paso, se prepara tu entorno de Cloud Shell, se configura tu proyecto de Google Cloud y se clona el repositorio inicial.

Abra Cloud Shell

Abre Cloud Shell en tu navegador. Cloud Shell proporciona un entorno preconfigurado con todas las herramientas que necesitas para este codelab. Haz clic en Autorizar cuando se te solicite

Luego, haz clic en "Ver" -> "Terminal" para abrir la terminal.Tu interfaz debería verse similar a esta:

Esta será nuestra interfaz principal, con el IDE en la parte superior y la terminal en la parte inferior.

Configura tu directorio de trabajo

Clona el repositorio de inicio. Todo el código que escribas en este codelab se encontrará aquí:

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

Crea el archivo .env a partir de la plantilla proporcionada:

cp .env.example .env

Para simplificar la configuración del proyecto en tu terminal, descarga esta secuencia de comandos de configuración del proyecto en tu directorio de trabajo:

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

Ejecuta la secuencia de comandos. Verifica tu cuenta de facturación de prueba, crea un proyecto nuevo (o valida uno existente), guarda el ID del proyecto en un archivo .env en el directorio actual y establece el proyecto activo en gcloud.

bash setup_verify_trial_project.sh && source .env

La secuencia de comandos hará lo siguiente:

Verifica que tengas una cuenta de facturación de prueba activa
Verifica si existe un proyecto en .env (si corresponde)
Crea un proyecto nuevo o reutiliza el existente
Vincula la cuenta de facturación de prueba a tu proyecto
Guarda el ID del proyecto en .env.
Configura el proyecto como el proyecto gcloud activo

Verifica que el proyecto esté configurado correctamente. Para ello, revisa el texto amarillo junto a tu directorio de trabajo en el mensaje de la terminal de Cloud Shell. Debería mostrar el ID de tu proyecto.

Activa la API requerida

A continuación, deberemos asegurarnos de habilitar la API necesaria para interactuar con Agent Platform de Gemini Enterprise

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

Configuración inicial de la infraestructura

Primero, deberemos instalar las dependencias de Python con uv, que es un administrador de proyectos y paquetes de Python rápido escrito en Rust ( documentación de uv). En este codelab, se usa por su velocidad y simplicidad para mantener el proyecto de Python.

uv sync

Luego, ejecuta la secuencia de comandos de configuración completa, que crea la instancia de Cloud SQL, inicializa los datos y, luego, implementa el servicio de Toolbox que actuará como el estado inicial de nuestro agente de restaurantes.

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

4. Concepto: Protocolo Agent2Agent (A2A) y entorno de ejecución de agentes de Gemini Enterprise

Antes de comenzar a compilar, dediquemos un momento a comprender las dos tecnologías clave que se presentan en este codelab para escalar nuestra aplicación basada en agentes.

El protocolo Agent2Agent (A2A)

El protocolo Agent2Agent (A2A) es un estándar abierto diseñado para permitir la comunicación y la colaboración fluidas entre agentes de IA. Mientras que el MCP (Protocolo de contexto del modelo) conecta agentes a herramientas y datos, el A2A conecta agentes a otros agentes, lo que les permite descubrir las capacidades de los demás, delegar tareas y colaborar en diferentes frameworks y organizaciones.

La diferencia clave entre encapsular un agente como una herramienta (a través de MCP) y exponerlo a través de A2A es que las herramientas no tienen estado y realizan funciones únicas, mientras que los agentes de A2A pueden razonar, mantener el estado y controlar interacciones de varios turnos, como negociaciones o aclaraciones. Un agente expuesto a través de A2A conserva todas sus capacidades en lugar de reducirse a una llamada a función.

A2A define tres conceptos fundamentales:

Tarjeta de agente: Es un documento JSON que describe lo que hace un agente, sus habilidades y su extremo. Otros agentes recuperan esta tarjeta para descubrir capacidades.
Mensaje: Es una solicitud del usuario o del agente que se envía a un extremo de A2A y que activa una tarea.
Tarea: Es una unidad de trabajo con un ciclo de vida (enviada → en proceso → completada/fallida) y artefactos que contienen los resultados.

Para obtener más información, consulta ¿Qué es A2A?

Tiempo de ejecución de Gemini Enterprise Agent Platform

Agent Runtime es un servicio completamente administrado en Google Cloud para implementar, escalar y administrar agentes de IA en producción con funciones de seguridad empresarial (p.ej., Controles del servicio de VPC y CMEK). Se encarga de la infraestructura para que puedas enfocarte en la lógica del agente.

Agent Runtime proporciona lo siguiente:

Implementación administrada: Implementa agentes creados con ADK, LangGraph o cualquier framework de Python con una sola llamada al SDK.
Alojamiento de A2A: Implementa agentes como extremos que cumplen con A2A con publicación automática de tarjetas de agente y acceso autenticado.
Sesiones persistentes: VertexAiSessionService almacena el historial y el estado de la conversación en todas las solicitudes.
Ajuste de escala automático: Se ajusta desde cero para controlar el tráfico, sin necesidad de administrar la infraestructura.
Observabilidad: Registro, supervisión y seguimiento integrados a través de la pila de observabilidad de Google Cloud
y muchas más funciones. Consulta esta documentación para obtener más detalles.

En este codelab, implementarás el agente de reservas en Agent Runtime. El proceso de implementación serializa (pickle) el código de tu agente y lo sube. El tiempo de ejecución del agente aprovisiona un extremo sin servidor que entrega el protocolo A2A. Otros agentes (o clientes) interactúan con él a través de llamadas HTTP estándar, autenticadas con credenciales de Google Cloud.

5. Compila el agente de reservas

En este paso, se crea un nuevo agente del ADK que controla las reservas de restaurantes con el estado de la sesión. El agente admite tres operaciones: crear, verificar y cancelar, con el número de teléfono como clave de búsqueda. Todos los datos de la reserva se encuentran en el estado de la sesión del ADK.

Crea el esqueleto del agente

Usa adk create para generar la estructura de directorios del agente con la configuración correcta del modelo y del proyecto:

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

Esto crea un directorio reservation_agent/ con __init__.py, agent.py y .env preconfigurados para el modelo de Gemini en Agent Platform.

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

A continuación, actualicemos el código del agente

Escribe el código del agente

Abre el archivo del agente generado:

cloudshell edit reservation_agent/agent.py

Luego, reemplaza el contenido con lo siguiente:

# 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. Prepara la configuración del servidor A2A

Define la tarjeta de agente A2A

La tarjeta de agente es una descripción estructurada de las capacidades de tu agente. Otros agentes y clientes la usan para descubrir qué hace tu agente. Crea la configuración de la tarjeta:

cloudshell edit reservation_agent/a2a_config.py

Copia lo siguiente en 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],
)

Crea el ejecutor de A2A

El ejecutor une el protocolo A2A y el agente del ADK. Recibe solicitudes de A2A, las ejecuta a través del agente del ADK y devuelve los resultados como tareas de A2A:

cloudshell edit reservation_agent/executor.py

Copia lo siguiente en 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())

El ejecutor detecta automáticamente su entorno: cuando se establece GOOGLE_CLOUD_AGENT_ENGINE_ID (Agent Runtime lo inyecta en el momento de la implementación), usa VertexAiSessionService para las sesiones persistentes. A nivel local, se recurre a InMemorySessionService.

Tu directorio reservation_agent ahora debería contener lo siguiente:

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

7. Cómo preparar un agente A2A con el SDK de Agent Platform y realizar pruebas locales

En este paso, se encapsula el agente de reservas como un agente compatible con A2A usando la clase A2aAgent del SDK de la plataforma de agentes ( el nombre del SDK aún usa el término vertex para la retrocompatibilidad) y, luego, se prueba el flujo completo del protocolo A2A de forma local: recuperación de la tarjeta de agente, envío de mensajes y recuperación de tareas. Este es el mismo objeto A2aAgent que implementarás en Agent Runtime en el siguiente paso.

Agrega dependencias

Instala el SDK de Agent Platform con compatibilidad con Agent Runtime y ADK, además del SDK de A2A:

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

Comprende los componentes de A2A

Para unir un agente del ADK para A2A, se requieren tres componentes:

Tarjeta de agente: Es una "tarjeta de presentación" que describe las capacidades, las habilidades y la URL del extremo del agente. Otros agentes usan esta información para descubrir qué hace tu agente.
Ejecutor de agentes: Es el puente entre el protocolo A2A y la lógica de tu agente del ADK. Recibe solicitudes de A2A, las ejecuta a través del agente del ADK y devuelve los resultados como tareas de A2A.
A2aAgent: Es la clase del SDK de Agent Platform que combina la tarjeta y el ejecutor en una unidad implementable.

Cómo crear la secuencia de comandos de prueba

Crea la siguiente secuencia de comandos para realizar pruebas de forma local

cloudshell edit scripts/test_a2a_agent_local.py

Copia lo siguiente en 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())

La secuencia de comandos de prueba importa la tarjeta del agente y el ejecutor que creaste en el paso anterior, sin duplicación. Se creará un objeto A2aAgent local, se simularán llamadas al protocolo A2A a través de solicitudes HTTP simuladas y se verificarán las tres operaciones de reserva.

Como no se configuró ningún GOOGLE_CLOUD_AGENT_ENGINE_ID de forma local, el ejecutor usa InMemorySessionService. Cuando se implementa en Agent Runtime, el mismo ejecutor cambia automáticamente a VertexAiSessionService para las sesiones persistentes.

Ejecuta la prueba

PYTHONPATH=. uv run python scripts/test_a2a_agent_local.py

El resultado muestra cinco etapas:

Tarjeta de agente: Recupera las capacidades y habilidades del agente.
Crear reserva: Reserva una mesa y devuelve una tarea con la confirmación.
Get task result: Recupera la tarea completada con la respuesta.
Verificar reserva: Busca la reserva por número de teléfono.
Cancelar reserva: Cancela la reserva y confirma la acción.

Ejemplo del resultado como se muestra a continuación

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

En este punto, verificaste que la tarjeta del agente de A2A describe las habilidades correctas, que las tres operaciones de reserva funcionan a través del flujo de tareas o mensajes del protocolo de A2A y que el estado persiste en los mensajes dentro del mismo contexto.

8. Implementa el agente de reservas en Agent Runtime

En este paso, se implementa el agente de reservas en el entorno de ejecución de Agent Platform de Gemini Enterprise, una plataforma sin servidores completamente administrada que aloja tu agente y lo expone como un extremo seguro de A2A. Después de la implementación, cualquier cliente autorizado puede descubrir el agente e interactuar con él a través de los extremos HTTP estándar de A2A.

Crea el bucket de etapa de pruebas

Crea un bucket de Cloud Storage para la etapa de pruebas del tiempo de ejecución del agente. Agent Runtime usa este bucket para subir el código y las dependencias de tu agente durante la implementación:

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

Crea la secuencia de comandos de implementación

A continuación, deberemos preparar la secuencia de comandos de implementación.

cloudshell edit scripts/deploy_a2a_agent_runtime.py

Copia lo siguiente en 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()

La secuencia de comandos de implementación importa los mismos agent_card y ReservationAgentExecutor que se usan en las pruebas locales, por lo que no hay duplicación de código. Agent Runtime serializa (pickle) el objeto A2aAgent junto con sus dependencias para la implementación. Al final de la secuencia de comandos de implementación, se escribirá el valor de RESERVATION_AGENT_RESOURCE_NAME en el archivo .env.

Implementar en Agent Runtime

Ejecuta la secuencia de comandos de implementación:

PYTHONPATH=. uv run python scripts/deploy_a2a_agent_runtime.py

La implementación tarda entre 3 y 5 minutos. La secuencia de comandos aprovisiona un endpoint sin servidores en Agent Runtime que aloja el agente de reservas. Después de la implementación correcta, verás un resultado similar al que se muestra a continuación.

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

Puedes ver el agente implementado en la consola de Cloud. Busca Agent Platform en la barra de búsqueda de la consola.

Luego, en la pestaña de la izquierda, coloca el cursor sobre Agents y selecciona Deployments.

Verás Reservation Agent en la lista de implementaciones, como se muestra a continuación.

Prueba el agente implementado

Ahora, ya puedes probar el agente implementado. Para ello, crea una secuencia de comandos de prueba:

cloudshell edit scripts/test_a2a_agent_runtime.py

Copia lo siguiente en 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())

Luego, ejecutemos la prueba.

source .env
uv run python scripts/test_a2a_agent_runtime.py

El resultado muestra la tarjeta del agente con la habilidad "Reservas de restaurantes", seguida de la tarea que se completa con una confirmación de reserva.

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

El agente de reservas ahora se ejecuta correctamente como un extremo A2A administrado en Agent Runtime.

9. Integra el agente de reservas A2A con el agente raíz de restaurantes

En este paso, se actualiza el agente de restaurante para que use el agente de reservas implementado como un agente secundario A2A remoto. El orquestador se ejecuta de forma local, mientras que el agente de reservas se ejecuta en Agent Runtime, una integración parcial que valida la conexión de A2A antes de la implementación completa.

Cómo resolver la URL de la tarjeta de agente de A2A

RemoteA2aAgent necesita la URL de la tarjeta del agente de reservas implementado para descubrir sus capacidades. Crea una secuencia de comandos que recupere esta URL del tiempo de ejecución del agente y la escriba en .env del agente del restaurante:

cloudshell edit scripts/resolve_agent_card_url.py

Copia lo siguiente en 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())

Ejecuta la secuencia de comandos para propagar el archivo .env con la URL de la tarjeta del agente.

uv run python scripts/resolve_agent_card_url.py
source .env

Actualiza el agente del restaurante

Abre el archivo del agente del restaurante:

cloudshell edit restaurant_agent/agent.py

Luego, reemplaza el contenido por la versión actualizada que incluye el agente de reservas remoto como agente secundario:

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

Los cambios clave con respecto a la versión anterior son los siguientes:

GoogleCloudAuth: Es un controlador httpx.Auth personalizado que actualiza el token de acceso de Google Cloud antes de cada solicitud. El entorno de ejecución del agente requiere llamadas de A2A autenticadas, y los tokens vencen después de un período.
RemoteA2aAgent lee RESERVATION_AGENT_CARD_URL del .env (escrito por el script de resolución) y usa el httpx_client autenticado.
Registrado como agente secundario: El organizador del ADK le delega automáticamente las solicitudes de reserva.
Se actualizó la instrucción para mencionar la delegación de reservas

Prueba el agente integrado de forma local

El agente inicial requería la integración con MCP Toolbox, y el archivo necesario ya debería haberse proporcionado en el codelab anterior o en el repo inicial. Solo debemos asegurarnos de que el proceso de la caja de herramientas se ejecute correctamente.

Si TOOLBOX_URL en tu .env ya apunta a un servicio de Cloud Run (del codelab anterior o tal vez del full_setup.sh del repo de inicio), puedes omitir este paso. El agente se conectará a la Toolbox implementada.

Si necesitas una Toolbox local, verifica si ya se está ejecutando una antes de iniciar una instancia nueva:

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

Luego, podemos intentar interactuar con el agente de restaurantes a través de la IU web de desarrollo del ADK.

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

Abre la IU web del ADK con la vista previa en la Web de Cloud Shell (haz clic en el botón Vista previa en la Web y cambia el puerto a 8080) y, luego, selecciona restaurant_agent.

Prueba una conversación mixta:

Consulta de menú

What Italian dishes do you have?

Solicitud de reserva

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

Verificar reserva

Crea una sesión nueva ( inicia una conversación desde cero):

Check the reservation for 123456

Detén el proceso adk web con Ctrl + C dos veces. A continuación, completaremos el sistema implementando por completo el agente.

10. Implementa el agente de restaurantes actualizado en Cloud Run

En este paso, se vuelve a implementar el agente de restaurantes en Cloud Run con la integración de A2A, lo que completa el sistema multiagente completamente implementado.

Otorgar permisos para acceder a Agent Runtime

La cuenta de servicio de Cloud Run necesita permiso para llamar a Agent Runtime. Otorga el rol roles/aiplatform.user a la cuenta de servicio predeterminada de 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"

Implementa en Cloud Run

En esta configuración, suponemos que el servicio del agente de restaurantes ya existe desde el codelab anterior o ejecutando scripts/full_setup.sh si comienzas desde cero. Esto se vuelve a implementar con el código actualizado (nueva integración de RemoteA2aAgent) y agrega la URL de la tarjeta del agente de reservas como una nueva variable de entorno. Se conservan las variables de entorno existentes (TOOLBOX_URL, GOOGLE_CLOUD_PROJECT, etc.):

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

Prueba el sistema completamente implementado

Obtén la URL del servicio implementado:

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

Abre la URL en tu navegador. Se carga la IU web del ADK, que es la misma interfaz que usaste de forma local y que ahora se ejecuta en Cloud Run.

No dudes en charlar con el agente.

Consulta de menú

What spicy dishes do you have?

Solicitud de reserva

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

Verificar reserva

Crea una sesión nueva ( inicia una conversación desde cero):

Check reservation for 555-0505

El sistema multiagente se implementó por completo. El agente de restaurantes en Cloud Run coordina dos servicios de backend: MCP Toolbox para las operaciones del menú y el agente de reservas A2A en Agent Runtime.

11. ¡Felicitaciones!

Creaste e implementaste un sistema multiagente con el protocolo A2A en Google Cloud.

Qué aprendiste

Compiló un agente del ADK que usa el estado de la sesión (ToolContext) para administrar los datos de reserva sin una base de datos.
Implementaste un agente A2A en Agent Runtime con el SDK de Agent Platform
Se consumió un agente A2A remoto de otro agente del ADK con RemoteA2aAgent como agente secundario.
Probamos el sistema de forma incremental: A2A local → A2A implementado → integración parcial → implementación completa

Realiza una limpieza

Para evitar que se apliquen cargos a tu cuenta de Google Cloud, borra los recursos que creaste en este codelab.

Opción 1: Borra el proyecto (recomendada)

gcloud projects delete $GOOGLE_CLOUD_PROJECT

Opción 2: Borra los recursos individuales

# 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