Se usó la API de Cloud Translation para traducir esta página.

Primeros pasos con el protocolo Agent-to-Agent (A2A): Interacciones entre un asistente de compras y un agente de ventas remoto con Gemini en Cloud Run y Agent Engine

1. Introducción

El protocolo de agente a agente (A2A) está diseñado para estandarizar la comunicación entre los agentes de IA, en particular para aquellos que se implementan en sistemas externos. Anteriormente, se establecieron protocolos similares para las Herramientas, denominados Protocolo de contexto del modelo (MCP), que es un estándar emergente para conectar LLMs con datos y recursos. A2A intenta complementar el MCP, ya que se enfoca en un problema diferente. Mientras que el MCP se enfoca en reducir la complejidad para conectar agentes con herramientas y datos, el A2A se enfoca en cómo permitir que los agentes colaboren en sus modalidades naturales. Permite que los agentes se comuniquen como agentes (o como usuarios) en lugar de como herramientas; por ejemplo, habilita la comunicación bidireccional cuando quieres pedir algo.

A2A se posiciona para complementar el MCP. En la documentación oficial, se recomienda que las aplicaciones usen el MCP para las herramientas y el A2A para los agentes, representados por AgentCard ( lo analizaremos más adelante). Luego, los frameworks pueden usar A2A para comunicarse con su usuario, los agentes remotos y otros agentes.

En esta demostración, comenzaremos con la implementación de A2A usando su SDK de Python. Exploraremos un caso de uso en el que tenemos un asistente personal de compras que puede ayudarnos a comunicarnos con los agentes de ventas de hamburguesas y pizzas para administrar nuestro pedido.

La A2A utiliza el principio cliente-servidor. Este es el flujo típico de A2A que esperamos en esta demostración

Primero, el cliente de A2A realizará la detección en todas las tarjetas de agente del servidor de A2A accesibles y utilizará su información para crear un cliente de conexión.
Cuando sea necesario, el cliente de A2A enviará un mensaje al servidor de A2A, que lo evaluará como una tarea que se debe completar. Si la URL del receptor de notificaciones push está configurada en el cliente de A2A y es compatible con el servidor de A2A, el servidor también podrá publicar el estado del progreso de la tarea en el extremo receptor del cliente.
Una vez que finalice la tarea, el servidor de A2A enviará el artefacto de respuesta al cliente de A2A.

En el codelab, seguirás un enfoque paso a paso de la siguiente manera:

Prepara tu proyecto de Google Cloud y habilita todas las APIs necesarias en él
Configura el espacio de trabajo para tu entorno de programación
Prepara variables de entorno para los servicios de agentes de hamburguesas y pizzas, y pruébalas de forma local
Implementa el agente de hamburguesas y pizzas en Cloud Run
Inspecciona los detalles sobre cómo se estableció el servidor de A2A
Prepara las variables de entorno para el asistente de compras y pruébalas de forma local
Implementa el asistente de compras en Agent Engine
Conéctate al motor del agente a través de la interfaz local
Inspecciona los detalles sobre cómo se estableció el cliente de A2A y su modelado de datos
Inspecciona la carga útil y la interacción entre el cliente y el servidor de A2A

Descripción general de la arquitectura

Implementaremos la siguiente arquitectura de servicio

Implementaremos 2 servicios que actuarán como servidor de A2A: el agente de hamburguesas ( respaldado por el framework de agentes de CrewAI) y el agente de pizzas ( respaldado por el framework de agentes de Langgraph). El usuario solo interactuará directamente con el asistente de compras, que se ejecutará con el framework del Kit de desarrollo de agentes (ADK) y actuará como cliente de A2A.

Cada uno de estos agentes tendrá su propio entorno y su propia implementación.

Requisitos previos

Comodidad para trabajar con Python
Conocimiento de la arquitectura básica de pila completa con el servicio HTTP

Qué aprenderás

Estructura principal del servidor de A2A
Estructura principal del cliente de A2A
Implementa el servicio del agente en Cloud Run
Implementa el servicio de agentes en Agent Engine
Cómo se conecta el cliente de A2A al servidor de A2A
Estructura de solicitud y respuesta en una conexión sin transmisión

Requisitos

Navegador web Chrome
Una cuenta de Gmail
Un proyecto de Cloud con la facturación habilitada

Este codelab, diseñado para desarrolladores de todos los niveles (incluidos los principiantes), usa Python en su aplicación de ejemplo. Sin embargo, no es necesario tener conocimientos de Python para comprender los conceptos que se presentan.

2. Antes de comenzar

Selecciona el proyecto activo en la consola de Cloud

En este codelab, se supone que ya tienes un proyecto de Google Cloud con la facturación habilitada. Si aún no lo tienes, puedes seguir las instrucciones que se indican a continuación para comenzar.

En la página del selector de proyectos de la consola de Google Cloud, selecciona o crea un proyecto de Google Cloud.
Asegúrate de que la facturación esté habilitada para tu proyecto de Cloud. Obtén información sobre cómo verificar si la facturación está habilitada en un proyecto.

Configura el proyecto de Cloud en la terminal de Cloud Shell

Usarás Cloud Shell, un entorno de línea de comandos que se ejecuta en Google Cloud y que viene precargado con bq. Haz clic en Activar Cloud Shell en la parte superior de la consola de Google Cloud. Si se te solicita autorización, haz clic en Autorizar.

Una vez que te conectes a Cloud Shell, verifica que ya te autenticaste y que el proyecto se configuró con el ID de tu proyecto con el siguiente comando:

gcloud auth list

En Cloud Shell, ejecuta el siguiente comando para confirmar que el comando gcloud conoce tu proyecto.

gcloud config list project

Si tu proyecto no está configurado, usa el siguiente comando para hacerlo:

gcloud config set project <YOUR_PROJECT_ID>

Como alternativa, también puedes ver el ID de PROJECT_ID en la consola.

Haz clic en él y verás todos tus proyectos y el ID del proyecto en el lado derecho.

Habilita las APIs requeridas con el siguiente comando. Este proceso puede tardar unos minutos, así que ten paciencia.

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

Si el comando se ejecuta correctamente, deberías ver un mensaje similar al que se muestra a continuación:

Operation "operations/..." finished successfully.

La alternativa al comando de gcloud es a través de la consola, buscando cada producto o usando este vínculo.

Si olvidas alguna API, siempre puedes habilitarla durante el curso de la implementación.

Consulta la documentación para ver los comandos y el uso de gcloud.

Ve al editor de Cloud Shell y configura el directorio de trabajo de la aplicación

Ahora, podemos configurar nuestro editor de código para hacer algunas cosas de programación. Usaremos el editor de Cloud Shell para esto.

Haz clic en el botón Abrir editor para abrir un editor de Cloud Shell en el que puedes escribir tu código
Asegúrate de que el proyecto de Cloud Code esté configurado en la esquina inferior izquierda (barra de estado) del editor de Cloud Shell, como se destaca en la siguiente imagen, y que esté configurado en el proyecto activo de Google Cloud en el que tienes habilitada la facturación. Haz clic en Autorizar si se te solicita. Si ya seguiste el comando anterior, es posible que el botón también apunte directamente a tu proyecto activado en lugar del botón de acceso.

A continuación, clonemos el directorio de trabajo de la plantilla para este codelab desde GitHub. Para ello, ejecuta el siguiente comando. Se creará el directorio de trabajo en el directorio purchasing-concierge-a2a.

git clone https://github.com/alphinside/purchasing-concierge-intro-a2a-codelab-starter.git purchasing-concierge-a2a

Después, ve a la sección superior del editor de Cloud Shell y haz clic en File->Open Folder, busca tu directorio de nombre de usuario y el directorio purchasing-concierge-a2a y, luego, haz clic en el botón Aceptar. Esto hará que el directorio elegido sea el directorio de trabajo principal. En este ejemplo, el nombre de usuario es alvinprayuda, por lo que la ruta de acceso del directorio se muestra a continuación.

Ahora, tu editor de Cloud Shell debería verse así:

Configuración del entorno

El siguiente paso es preparar el entorno de desarrollo. Tu terminal activa actual debe estar dentro del directorio de trabajo purchasing-concierge-a2a. En este codelab, usaremos Python 3.12 y uv python project manager para simplificar la necesidad de crear y administrar la versión de Python y el entorno virtual.

Si aún no abriste la terminal, haz clic en Terminal -> New Terminal o usa Ctrl + Mayúsculas + C para abrir una ventana de terminal en la parte inferior del navegador.

Ahora, inicialicemos el entorno virtual del asistente de compras con uv (ya preinstalado en la terminal de Cloud). Ejecuta este comando:

uv sync --frozen

Esto creará el directorio .venv y, luego, instalará las dependencias. Un vistazo rápido a pyproject.toml te brindará información sobre las dependencias que se muestran de esta manera:

dependencies = [
    "a2a-sdk>=0.2.16",
    "google-adk>=1.8.0",
    "gradio>=5.38.2",
]

Para probar el entorno virtual, crea un archivo nuevo main.py y copia el siguiente código:

def main():
   print("Hello from purchasing-concierge-a2a!")

if __name__ == "__main__":
   main()

Luego, ejecuta el siguiente comando:

uv run main.py

Obtendrás un resultado como el que se muestra a continuación.

Using CPython 3.12
Creating virtual environment at: .venv
Hello from purchasing-concierge-a2a!

Esto demuestra que el proyecto de Python se está configurando correctamente.

Ahora podemos pasar al siguiente paso, que consiste en configurar y, luego, implementar nuestro agente de vendedor remoto.

3. Implementación del agente de vendedor remoto: servidor de A2A en Cloud Run

En este paso, implementaremos estos dos agentes de vendedores remotos marcados con el cuadro rojo. El agente de hamburguesas estará potenciado por el framework de agentes de CrewAI y el agente de pizzas estará potenciado por el agente de Langgraph. Ambos estarán potenciados por el modelo Gemini Flash 2.0.

Implementa el agente de hamburguesas remoto

El código fuente del agente de hamburguesas se encuentra en el directorio remote_seller_agents/burger_agent. La inicialización del agente se puede inspeccionar en la secuencia de comandos agent.py. Este es el fragmento de código del agente inicializado

from crewai import Agent, Crew, LLM, Task, Process
from crewai.tools import tool

...

       model = LLM(
            model="vertex_ai/gemini-2.5-flash-lite",  # Use base model name without provider prefix
        )
        burger_agent = Agent(
            role="Burger Seller Agent",
            goal=(
                "Help user to understand what is available on burger menu and price also handle order creation."
            ),
            backstory=("You are an expert and helpful burger seller agent."),
            verbose=False,
            allow_delegation=False,
            tools=[create_burger_order],
            llm=model,
        )

        agent_task = Task(
            description=self.TaskInstruction,
            agent=burger_agent,
            expected_output="Response to the user in friendly and helpful manner",
        )

        crew = Crew(
            tasks=[agent_task],
            agents=[burger_agent],
            verbose=False,
            process=Process.sequential,
        )

        inputs = {"user_prompt": query, "session_id": sessionId}
        response = crew.kickoff(inputs)
        return response

...

Todos los archivos que existen en el directorio remote_seller_agents/burger_agent ya son suficientes para implementar nuestro agente en Cloud Run de modo que se pueda acceder a él como un servicio. Hablaremos de esto más adelante. Ejecuta el siguiente comando para implementarlo:

gcloud run deploy burger-agent \
    --source remote_seller_agents/burger_agent \
    --port=8080 \
    --allow-unauthenticated \
    --min 1 \
    --region us-central1 \
    --update-env-vars GOOGLE_CLOUD_LOCATION=us-central1 \
    --update-env-vars GOOGLE_CLOUD_PROJECT={your-project-id}

Si se te solicita que se cree un repositorio de contenedores para la implementación desde la fuente, responde Y. Esto solo ocurrió si nunca antes implementaste un Cloud Run desde la fuente. Después de la implementación correcta, se mostrará un registro como este.

Service [burger-agent] revision [burger-agent-xxxxx-xxx] has been deployed and is serving 100 percent of traffic.
Service URL: https://burger-agent-xxxxxxxxx.us-central1.run.app

La parte xxxx aquí será un identificador único cuando implementemos el servicio. Ahora, intentemos acceder a la ruta https://burger-agent-xxxxxxxxx.us-central1.run.app/.well-known/agent.json de esos servicios de agentes de hamburguesas implementados a través del navegador. Esta es la URL para acceder a la tarjeta del agente del servidor de A2A implementado.

Si la implementación se realiza correctamente, verás la respuesta que se muestra a continuación en tu navegador cuando accedas a https://burger-agent-xxxxxxxxx.us-central1.run.app/.well-known/agent.json :

Esta es la información de la tarjeta del agente de hamburguesas a la que se debe poder acceder para fines de descubrimiento. Hablaremos de esto más adelante. Observa que el valor de url aún está establecido en http://0.0.0.0:8080/ aquí. Este valor de url debería ser la información principal para que el cliente de A2A envíe mensajes desde el mundo exterior, pero no está configurado correctamente. Para esta demostración, deberemos actualizar este valor a la URL de nuestro servicio de agente de hamburguesas agregando una variable de entorno adicional HOST_OVERRIDE.

Cómo actualizar el valor de la URL del agente de hamburguesas en la tarjeta del agente a través de la variable de entorno

Para agregar HOST_OVERRIDE al servicio de agente de hamburguesas, sigue estos pasos:

Busca Cloud Run en la barra de búsqueda que se encuentra en la parte superior de la consola de Cloud.

Haz clic en el servicio de Cloud Run burger-agent implementado anteriormente.

Copia la URL de burger-service y, luego, haz clic en Editar e implementar nueva revisión.

Luego, haz clic en la sección Variables y Secrets.

Después, haz clic en Agregar variable y establece el valor de HOST_OVERRIDE en la URL del servicio ( la que tiene el patrón https://burger-agent-xxxxxxxxx.us-central1.run.app).

Por último, haz clic en el botón deploy para volver a implementar tu servicio.

Ahora, cuando vuelvas a acceder a la tarjeta del agente burger-agent en el navegador https://burger-agent-xxxxxxxxx.us-central1.run.app/.well-known/agent.json , el valor de url ya estará configurado correctamente.

Implementación del agente de Remote Pizza

Del mismo modo, el código fuente del agente de pizza se encuentra en el directorio remote_seller_agents/pizza_agent. La inicialización del agente se puede inspeccionar en la secuencia de comandos agent.py. Este es el fragmento de código del agente inicializado

from langchain_google_vertexai import ChatVertexAI
from langgraph.prebuilt import create_react_agent

...

self.model = ChatVertexAI(
    model="gemini-2.5-flash-lite",
    location=os.getenv("GOOGLE_CLOUD_LOCATION"),
    project=os.getenv("GOOGLE_CLOUD_PROJECT"),
)
self.tools = [create_pizza_order]
self.graph = create_react_agent(
    self.model,
    tools=self.tools,
    checkpointer=memory,
    prompt=self.SYSTEM_INSTRUCTION,
)

...

Al igual que en el paso anterior de implementación del agente de hamburguesas, todos los archivos que existen en el directorio remote_seller_agents/pizza_agent ya son suficientes para implementar nuestro agente en Cloud Run de modo que se pueda acceder a él como un servicio. Ejecuta el siguiente comando para implementarlo:

gcloud run deploy pizza-agent \
    --source remote_seller_agents/pizza_agent \
    --port=8080 \
    --allow-unauthenticated \
    --min 1 \
    --region us-central1 \
    --update-env-vars GOOGLE_CLOUD_LOCATION=us-central1 \
    --update-env-vars GOOGLE_CLOUD_PROJECT={your-project-id}

Después de la implementación correcta, se mostrará un registro como este.

Service [pizza-agent] revision [pizza-agent-xxxxx-xxx] has been deployed and is serving 100 percent of traffic.
Service URL: https://pizza-agent-xxxxxxxxx.us-central1.run.app

La parte xxxx aquí será un identificador único cuando implementemos el servicio. Lo mismo sucede con el agente de hamburguesas. Cuando intentas acceder a la ruta https://pizza-agent-xxxxxxxxx.us-central1.run.app/.well-known/agent.json de esos servicios del agente de pizzas implementados a través del navegador para acceder a la tarjeta del agente del servidor de A2A, el valor url del agente de pizzas en su tarjeta de agente aún no está configurado correctamente. También debemos agregar HOST_OVERRIDE a su variable de entorno.

Actualiza el valor de la URL del agente de pizza en la tarjeta del agente a través de la variable de entorno

Para agregar HOST_OVERRIDE al servicio de agente de pizza, sigue estos pasos:

Busca Cloud Run en la barra de búsqueda que se encuentra en la parte superior de la consola de Cloud.

Haz clic en el servicio de Cloud Run pizza-agent implementado anteriormente.

Haz clic en Editar e implementar nueva revisión.

Copia la URL de pizza-service y, luego, haz clic en la sección Variables y Secrets.

Después, haz clic en Agregar variable y establece el valor de HOST_OVERRIDE en la URL del servicio ( la que tiene el patrón https://pizza-agent-xxxxxxxxx.us-central1.run.app).

Por último, haz clic en el botón deploy para volver a implementar tu servicio.

Ahora, cuando vuelvas a acceder a la tarjeta del agente pizza-agent en el navegador https://pizza-agent-xxxxxxxxx.us-central1.run.app/.well-known/agent.json, el valor url ya estará configurado correctamente.

En este punto, ya implementamos correctamente los servicios de hamburguesas y pizzas en Cloud Run. Ahora, analicemos los componentes principales del servidor de A2A

4. Componentes principales del servidor de A2A

Ahora hablemos del concepto y los componentes principales del servidor de A2A

Tarjeta de agente

Cada servidor de A2A debe tener una tarjeta de agente a la que se pueda acceder en el recurso /.well-known/agent.json. Esto es para respaldar la fase de descubrimiento en el cliente de A2A, que debe proporcionar información y contextos completos sobre cómo acceder al agente y conocer todas sus capacidades. Es algo similar a la documentación de la API bien documentada con Swagger o Postman.

Este es el contenido de la tarjeta del agente de hamburguesas implementado.

{
  "capabilities": {
    "streaming": true
  },
  "defaultInputModes": [
    "text",
    "text/plain"
  ],
  "defaultOutputModes": [
    "text",
    "text/plain"
  ],
  "description": "Helps with creating burger orders",
  "name": "burger_seller_agent",
  "protocolVersion": "0.2.6",
  "skills": [
    {
      "description": "Helps with creating burger orders",
      "examples": [
        "I want to order 2 classic cheeseburgers"
      ],
      "id": "create_burger_order",
      "name": "Burger Order Creation Tool",
      "tags": [
        "burger order creation"
      ]
    }
  ],
  "url": "https://burger-agent-109790610330.us-central1.run.app",
  "version": "1.0.0"
}

Estas tarjetas de agentes destacan muchos componentes importantes, como las habilidades del agente, las capacidades de transmisión, las modalidades admitidas, la versión del protocolo y otros aspectos.

Toda esta información se puede utilizar para desarrollar un mecanismo de comunicación adecuado para que el cliente de A2A se pueda comunicar correctamente. La modalidad y el mecanismo de autenticación admitidos garantizan que la comunicación se pueda establecer correctamente y que la información del agente skills se pueda incorporar en la instrucción del sistema del cliente de A2A para brindarle contexto al agente del cliente sobre las capacidades y habilidades del agente remoto que se invocarán. Puedes encontrar campos más detallados para esta tarjeta de agente en esta documentación.

En nuestro código, la implementación de la tarjeta del agente se establece con el SDK de Python de A2A. Consulta el siguiente fragmento de remote_seller_agents/burger_agent/main.py para ver la implementación.

...

        capabilities = AgentCapabilities(streaming=True)
        skill = AgentSkill(
            id="create_burger_order",
            name="Burger Order Creation Tool",
            description="Helps with creating burger orders",
            tags=["burger order creation"],
            examples=["I want to order 2 classic cheeseburgers"],
        )
        agent_host_url = (
            os.getenv("HOST_OVERRIDE")
            if os.getenv("HOST_OVERRIDE")
            else f"http://{host}:{port}/"
        )
        agent_card = AgentCard(
            name="burger_seller_agent",
            description="Helps with creating burger orders",
            url=agent_host_url,
            version="1.0.0",
            defaultInputModes=BurgerSellerAgent.SUPPORTED_CONTENT_TYPES,
            defaultOutputModes=BurgerSellerAgent.SUPPORTED_CONTENT_TYPES,
            capabilities=capabilities,
            skills=[skill],
        )

...

Allí, podemos ver varios campos, como los siguientes:

AgentCapabilities : Declaración de funciones opcionales adicionales que admite el servicio del agente,como la capacidad de transmisión o la compatibilidad con notificaciones push
AgentSkill : Herramientas o funciones compatibles con el agente
Input/OutputModes : Modalidad de tipo de entrada/salida admitida
Url : Dirección para comunicarse con el agente

En esta configuración, proporcionamos una creación dinámica de la URL del host del agente, de modo que sea más fácil cambiar entre las pruebas locales y la implementación en la nube. Por eso, necesitamos agregar la variable HOST_OVERRIDE en el paso anterior.

Task Queue y Agent Executor

El servidor de A2A puede controlar solicitudes de diferentes agentes o usuarios, y aislar cada tarea a la perfección. Para visualizar mejor los contextos de estos, puedes inspeccionar la siguiente imagen.

Por lo tanto, cada servidor de A2A debe poder hacer un seguimiento de las tareas entrantes y almacenar la información adecuada sobre ellas. El SDK de A2A proporciona módulos para abordar este desafío en el servidor de A2A. Primero, podemos crear una instancia de la lógica sobre cómo queremos controlar la solicitud entrante. Si heredamos la clase abstracta AgentExecutor, podemos controlar cómo queremos administrar la ejecución y la cancelación de tareas. Esta implementación de ejemplo se puede inspeccionar en el módulo remote_seller_agents/burger_agent/agent_executor.py ( ruta similar para el caso del vendedor de pizza).

...

class BurgerSellerAgentExecutor(AgentExecutor):
    """Burger Seller AgentExecutor."""

    def __init__(self):
        self.agent = BurgerSellerAgent()

    async def execute(
        self,
        context: RequestContext,
        event_queue: EventQueue,
    ) -> None:
        query = context.get_user_input()
        try:
            result = self.agent.invoke(query, context.context_id)
            print(f"Final Result ===> {result}")

            parts = [Part(root=TextPart(text=str(result)))]
            await event_queue.enqueue_event(
                completed_task(
                    context.task_id,
                    context.context_id,
                    [new_artifact(parts, f"burger_{context.task_id}")],
                    [context.message],
                )
            )
        except Exception as e:
            print("Error invoking agent: %s", e)
            raise ServerError(error=ValueError(f"Error invoking agent: {e}")) from e

    async def cancel(
        self, request: RequestContext, event_queue: EventQueue
    ) -> Task | None:
        raise ServerError(error=UnsupportedOperationError())

...

En el código anterior, implementamos un esquema de procesamiento básico en el que se invocará directamente al agente cuando llegue la solicitud y se enviarán eventos de tareas completadas después de que finalice la invocación. Sin embargo, no implementamos el método de cancelación aquí, ya que se consideró una operación de corta duración.

Después de compilar el ejecutor, podemos utilizar directamente los elementos integrados DefaultRequestHandler, InMemoryTaskStore y A2AStarletteApplication para iniciar el servidor HTTP. Esta implementación se puede inspeccionar en remote_seller_agents/burger_agent/__main__.py.

...

        request_handler = DefaultRequestHandler(
            agent_executor=BurgerSellerAgentExecutor(),
            task_store=InMemoryTaskStore(),
        )
        server = A2AStarletteApplication(
            agent_card=agent_card, http_handler=request_handler
        )

        uvicorn.run(server.build(), host=host, port=port)

...

En este módulo, se te proporcionará la implementación de la ruta /.well-known/agent.json para acceder a la tarjeta del agente y también el extremo POST para admitir el protocolo A2A.

Resumen

En resumen, hasta ahora, nuestro servidor de A2A implementado usa el SDK de Python que puede admitir las 2 funcionalidades que se indican a continuación:

Publicar la tarjeta del agente en la ruta del /.well-known/agent.json
Controla la solicitud de JSON-RPC con la cola de tareas en memoria

El punto de entrada para iniciar estas funcionalidades se puede inspeccionar en la secuencia de comandos __main__.py ( en remote_seller_agents/burger_agent o remote_seller_agents/pizza_agent) .

5. Implementación del Concierge de compras: cliente de A2A en Agent Engine

En este paso, implementaremos el agente de asistente de compras. Este es el agente con el que interactuaremos.

El código fuente de nuestro agente de asistente de compras se encuentra en el directorio purchasing_concierge. La inicialización del agente se puede inspeccionar en la secuencia de comandos purchasing_agent.py. Este es el fragmento de código del agente inicializado.

from google.adk import Agent

...

def create_agent(self) -> Agent:
        return Agent(
            model="gemini-2.5-flash-lite",
            name="purchasing_agent",
            instruction=self.root_instruction,
            before_model_callback=self.before_model_callback,
            before_agent_callback=self.before_agent_callback,
            description=(
                "This purchasing agent orchestrates the decomposition of the user purchase request into"
                " tasks that can be performed by the seller agents."
            ),
            tools=[
                self.send_task,
            ],
        )

...

Implementaremos este agente en Agent Engine. Vertex AI Agent Engine es un conjunto de servicios que permite a los desarrolladores implementar, administrar y escalar agentes de IA en producción. Administra la infraestructura para escalar agentes en producción, de modo que podamos enfocarnos en crear aplicaciones. Puedes obtener más información al respecto en este documento . Si antes necesitábamos preparar los archivos necesarios para implementar nuestro servicio de agente (como el script del servidor main y el Dockerfile), en este caso podemos implementar nuestro agente directamente desde el script de Python sin necesidad de desarrollar nuestro propio servicio de backend usando una combinación del ADK y Agent Engine. Sigue estos pasos para implementarlo :

Primero, debemos crear nuestro almacenamiento de etapa de pruebas en Cloud Storage.

gcloud storage buckets create gs://purchasing-concierge-{your-project-id} --location=us-central1

Ahora, primero debemos preparar la variable .env. Copiemos .env.example en el archivo .env.

cp .env.example .env

Ahora, abre el archivo .env y verás el siguiente contenido:

GOOGLE_GENAI_USE_VERTEXAI=TRUE
GOOGLE_CLOUD_PROJECT={your-project-id}
GOOGLE_CLOUD_LOCATION=us-central1
STAGING_BUCKET=gs://purchasing-concierge-{your-project-id}
PIZZA_SELLER_AGENT_URL={your-pizza-agent-url}
BURGER_SELLER_AGENT_URL={your-burger-agent-url}
AGENT_ENGINE_RESOURCE_NAME={your-agent-engine-resource-name}

Este agente se comunicará con el agente de hamburguesas y pizzas, por lo que debemos proporcionar las credenciales adecuadas para ambos. Deberemos actualizar PIZZA_SELLER_AGENT_URL y BURGER_SELLER_AGENT_URL con la URL de Cloud Run de los pasos anteriores.

Si lo olvidas, visitemos la consola de Cloud Run. Escribe "Cloud Run" en la barra de búsqueda que se encuentra en la parte superior de la consola y haz clic con el botón derecho en el ícono de Cloud Run para abrirlo en una pestaña nueva.

Deberías ver los servicios de agentes de vendedores remotos que implementamos anteriormente, como se muestra a continuación.

Ahora, para ver la URL pública de esos servicios, haz clic en uno de ellos y se te redireccionará a la página Detalles del servicio. Puedes ver la URL en el área superior, justo al lado de la información de la región.

La variable de entorno final debería ser similar a esta:

GOOGLE_GENAI_USE_VERTEXAI=TRUE
GOOGLE_CLOUD_PROJECT={your-project-id}
GOOGLE_CLOUD_LOCATION=us-central1
STAGING_BUCKET=gs://purchasing-concierge-{your-project-id}
PIZZA_SELLER_AGENT_URL=https://pizza-agent-xxxxx.us-central1.run.app
BURGER_SELLER_AGENT_URL=https://burger-agent-xxxxx.us-central1.run.app
AGENT_ENGINE_RESOURCE_NAME={your-agent-engine-resource-name}

Ahora, ya podemos implementar nuestro agente de asistente de compras. En esta demostración, realizaremos la implementación con la secuencia de comandos deploy_to_agent_engine.py, cuyo contenido se muestra a continuación.

"""
Copyright 2025 Google LLC

Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at

    https://www.apache.org/licenses/LICENSE-2.0

Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
"""

import vertexai
from vertexai.preview import reasoning_engines
from vertexai import agent_engines
from dotenv import load_dotenv
import os
from purchasing_concierge.agent import root_agent

load_dotenv()

PROJECT_ID = os.getenv("GOOGLE_CLOUD_PROJECT")
LOCATION = os.getenv("GOOGLE_CLOUD_LOCATION")
STAGING_BUCKET = os.getenv("STAGING_BUCKET")

vertexai.init(
    project=PROJECT_ID,
    location=LOCATION,
    staging_bucket=STAGING_BUCKET,
)

adk_app = reasoning_engines.AdkApp(
    agent=root_agent,
)

remote_app = agent_engines.create(
    agent_engine=adk_app,
    display_name="purchasing-concierge",
    requirements=[
        "google-cloud-aiplatform[adk,agent_engines]",
        "a2a-sdk==0.2.16",
    ],
    extra_packages=[
        "./purchasing_concierge",
    ],
    env_vars={
        "GOOGLE_GENAI_USE_VERTEXAI": os.environ["GOOGLE_GENAI_USE_VERTEXAI"],
        "PIZZA_SELLER_AGENT_URL": os.environ["PIZZA_SELLER_AGENT_URL"],
        "BURGER_SELLER_AGENT_URL": os.environ["BURGER_SELLER_AGENT_URL"],
    },
)

print(f"Deployed remote app resource: {remote_app.resource_name}")

Estos son los pasos necesarios para implementar nuestro agente de ADK en el motor de agentes. Primero, debemos crear un objeto AdkApp a partir de nuestro root_agent del ADK. Luego, podemos ejecutar el método agent_engines.create proporcionando el objeto adk_app, especificando los requisitos en el campo requirements, especificando la ruta de acceso del directorio del agente en extra_packages y proporcionando las variables de entorno necesarias.

Para implementarlo, ejecuta la secuencia de comandos:

uv run deploy_to_agent_engine.py

Después de la implementación correcta, se mostrará un registro como este. Ten en cuenta que xxxx es el ID de tu proyecto y yyyy es el ID del recurso del motor del agente.

AgentEngine created. Resource name: projects/xxxx/locations/us-central1/reasoningEngines/yyyy
To use this AgentEngine in another session:
agent_engine = vertexai.agent_engines.get('projects/xxxx/locations/us-central1/reasoningEngines/yyyy)
Deployed remote app resource: projects/xxxx/locations/us-central1/reasoningEngines/xxxx

Y cuando lo inspeccionemos en el panel del motor del agente (busca "motor del agente" en la barra de búsqueda), se mostrará nuestra implementación anterior.

Prueba el agente implementado en Agent Engine

La interacción con el motor de agentes se puede realizar a través del comando curl y el SDK. Por ejemplo, ejecuta el siguiente comando para intentar interactuar con el agente implementado.

Puedes intentar enviar esta consulta para verificar si el agente se implementó correctamente.

curl \
-H "Authorization: Bearer $(gcloud auth print-access-token)" \
-H "Content-Type: application/json" \
https://us-central1-aiplatform.googleapis.com/v1/projects/{YOUR_PROJECT_ID}/locations/us-central1/reasoningEngines/{YOUR_AGENT_ENGINE_RESOURCE_ID}:streamQuery?alt=sse -d '{
  "class_method": "stream_query",
  "input": {
    "user_id": "user_123",
    "message": "List available burger menu please",
  }
}'

Si se ejecuta de forma correcta, se mostrarán varios eventos de respuesta transmitidos en tu consola, como este:

{
  "content": {
    "parts": [
      {
        "text": "Here is our burger menu:\n- Classic Cheeseburger: IDR 85K\n- Double Cheeseburger: IDR 110K\n- Spicy Chicken Burger: IDR 80K\n- Spicy Cajun Burger: IDR 85K"
      }
    ],
    "role": "model"
  },
  "usage_metadata": {
    "candidates_token_count": 51,
    "candidates_tokens_details": [
      {
        "modality": "TEXT",
        "token_count": 51
      }
    ],
    "prompt_token_count": 907,
    "prompt_tokens_details": [
      {
        "modality": "TEXT",
        "token_count": 907
      }
    ],
    "total_token_count": 958,
    "traffic_type": "ON_DEMAND"
  },
  "invocation_id": "e-14679918-af68-45f1-b942-cf014368a733",
  "author": "purchasing_agent",
  "actions": {
    "state_delta": {},
    "artifact_delta": {},
    "requested_auth_configs": {}
  },
  "id": "dbe7fc43-b82a-4f3e-82aa-dd97afa8f15b",
  "timestamp": 1754287348.941454
}

En el siguiente paso, intentaremos usar la IU. Sin embargo, primero analicemos cuáles son los componentes principales y el flujo típico de los clientes de A2A.

6. Componentes principales del cliente de A2A

La imagen anterior muestra el flujo típico de las interacciones de A2A:

El cliente intentará encontrar cualquier tarjeta de agente publicada en la URL del agente remoto proporcionada en la ruta /.well-known/agent.json.
Luego, cuando sea necesario, enviará un mensaje a ese agente con el mensaje y los parámetros de metadatos necesarios ( p. ej., ID de sesión, contexto histórico, etcétera). El servidor percibirá este mensaje como una tarea que se debe completar.
El servidor de A2A procesa la solicitud y, si admite notificaciones push, también podrá publicar algunas notificaciones durante el procesamiento de la tarea ( esta funcionalidad está fuera del alcance de este codelab).
Una vez que finalice, el servidor de A2A enviará el artefacto de respuesta al cliente.

Algunos de los objetos principales para las interacciones anteriores son estos elementos (puedes leer más detalles aquí) :

Mensaje: Es un turno de comunicación entre un cliente y un agente remoto.
Task: Es la unidad de trabajo fundamental que administra A2A y que se identifica con un ID único.
Artefacto: Es un resultado (p. ej., un documento, una imagen o datos estructurados) que genera el agente como resultado de una tarea y que se compone de partes.
Parte: Es la unidad de contenido más pequeña dentro de un mensaje o artefacto. La parte puede ser un texto, una imagen, un video, un archivo, etcétera.

Descubrimiento de tarjetas

Cuando se inicia el servicio de cliente de A2A, el proceso típico es intentar obtener la información de la tarjeta del agente y almacenarla para acceder a ella fácilmente cuando sea necesario. En este codelab, la implementamos en before_agent_callback. Puedes ver la implementación en purchasing_concierge/purchasing_agent.py en el siguiente fragmento de código.

...

async def before_agent_callback(self, callback_context: CallbackContext):
        if not self.a2a_client_init_status:
            httpx_client = httpx.AsyncClient(timeout=httpx.Timeout(timeout=30))
            for address in self.remote_agent_addresses:
                card_resolver = A2ACardResolver(
                    base_url=address, httpx_client=httpx_client
                )
                try:
                    card = await card_resolver.get_agent_card()
                    remote_connection = RemoteAgentConnections(
                        agent_card=card, agent_url=card.url
                    )
                    self.remote_agent_connections[card.name] = remote_connection
                    self.cards[card.name] = card
                except httpx.ConnectError:
                    print(f"ERROR: Failed to get agent card from : {address}")
            agent_info = []
            for ra in self.list_remote_agents():
                agent_info.append(json.dumps(ra))
            self.agents = "\n".join(agent_info)

...

Aquí, intentamos acceder a todas las tarjetas de agentes disponibles con el módulo A2ACardResolver del cliente A2A integrado. Luego, recopilamos la conexión necesaria para enviar mensajes al agente. Después de eso, también debemos enumerar todos los agentes disponibles y sus especificaciones en la instrucción para que nuestro agente sepa que puede comunicarse con estos agentes.

Herramienta de instrucciones y envío de tareas

Esta es la instrucción y la herramienta que proporcionamos a nuestro agente del ADK aquí.

...

def root_instruction(self, context: ReadonlyContext) -> str:
    current_agent = self.check_active_agent(context)
    return f"""You are an expert purchasing delegator that can delegate the user product inquiry and purchase request to the
appropriate seller remote agents.

Execution:
- For actionable tasks, you can use `send_task` to assign tasks to remote agents to perform.
- When the remote agent is repeatedly asking for user confirmation, assume that the remote agent doesn't have access to user's conversation context. 
So improve the task description to include all the necessary information related to that agent
- Never ask user permission when you want to connect with remote agents. If you need to make connection with multiple remote agents, directly
connect with them without asking user permission or asking user preference
- Always show the detailed response information from the seller agent and propagate it properly to the user. 
- If the remote seller is asking for confirmation, rely the confirmation question to the user if the user haven't do so. 
- If the user already confirmed the related order in the past conversation history, you can confirm on behalf of the user
- Do not give irrelevant context to remote seller agent. For example, ordered pizza item is not relevant for the burger seller agent
- Never ask order confirmation to the remote seller agent 

Please rely on tools to address the request, and don't make up the response. If you are not sure, please ask the user for more details.
Focus on the most recent parts of the conversation primarily.

If there is an active agent, send the request to that agent with the update task tool.

Agents:
{self.agents}

Current active seller agent: {current_agent["active_agent"]}
"""

...

async def send_task(self, agent_name: str, task: str, tool_context: ToolContext):
        """Sends a task to remote seller agent

        This will send a message to the remote agent named agent_name.

        Args:
            agent_name: The name of the agent to send the task to.
            task: The comprehensive conversation context summary
                and goal to be achieved regarding user inquiry and purchase request.
            tool_context: The tool context this method runs in.

        Yields:
            A dictionary of JSON data.
        """
        if agent_name not in self.remote_agent_connections:
            raise ValueError(f"Agent {agent_name} not found")
        state = tool_context.state
        state["active_agent"] = agent_name
        client = self.remote_agent_connections[agent_name]
        if not client:
            raise ValueError(f"Client not available for {agent_name}")
        session_id = state["session_id"]
        task: Task
        message_id = ""
        metadata = {}
        if "input_message_metadata" in state:
            metadata.update(**state["input_message_metadata"])
            if "message_id" in state["input_message_metadata"]:
                message_id = state["input_message_metadata"]["message_id"]
        if not message_id:
            message_id = str(uuid.uuid4())

        payload = {
            "message": {
                "role": "user",
                "parts": [
                    {"type": "text", "text": task}
                ],  # Use the 'task' argument here
                "messageId": message_id,
                "contextId": session_id,
            },
        }

        message_request = SendMessageRequest(
            id=message_id, params=MessageSendParams.model_validate(payload)
        )
        send_response: SendMessageResponse = await client.send_message(
            message_request=message_request
        )
        print(
            "send_response",
            send_response.model_dump_json(exclude_none=True, indent=2),
        )

        if not isinstance(send_response.root, SendMessageSuccessResponse):
            print("received non-success response. Aborting get task ")
            return None

        if not isinstance(send_response.root.result, Task):
            print("received non-task response. Aborting get task ")
            return None

        return send_response.root.result

...

En la instrucción, le proporcionamos a nuestro agente de asistencia para compras todos los nombres y las descripciones de los agentes remotos disponibles, y en la herramienta self.send_task proporcionamos un mecanismo para recuperar el cliente adecuado para conectarse con el agente y enviar los metadatos requeridos con el objeto SendMessageRequest.

Los protocolos de comunicación

La definición de Task es un dominio que pertenece al servidor de A2A. Sin embargo, desde la perspectiva del cliente de A2A, lo ve como un Message que se envía al servidor. Depende del servidor cómo definir los mensajes entrantes del cliente como qué tarea y si completar la tarea necesita la interacción del cliente. Puedes leer más detalles sobre el ciclo de vida de la tarea en esta documentación. El concepto de nivel superior se puede visualizar a continuación:

Este intercambio de mensaje -> tarea se implementa con el formato de carga útil sobre el estándar JSON-RPC, como se muestra en el siguiente ejemplo del protocolo message/send :

{
  # identifier for this request
  "id": "abc123",
  # version of JSON-RPC protocol
  "jsonrpc": "2.0",
  # method name
  "method": "message/send",
  # parameters/arguments of the method
  "params": {
    "message": "hi, what can you help me with?"
  }  
}

Hay varios métodos disponibles, por ejemplo, para admitir diferentes tipos de comunicación (p.ej., sincrónica, transmisión, asincrónica) o para configurar notificaciones sobre el estado de la tarea. El servidor de A2A se puede configurar de forma flexible para controlar estos estándares de definición de tareas. Puedes leer los detalles de estos métodos en este documento.

7. Pruebas de integración y análisis de cargas útiles

Ahora, inspeccionemos nuestro asistente de compras con la interacción del agente remoto a través de una IU web.

Primero, debemos actualizar el AGENT_ENGINE_RESOURCE_NAME en el .env o en niveles inferiores. Asegúrate de proporcionar el nombre de recurso del motor del agente correcto. Tu archivo .env debería tener el siguiente aspecto:

GOOGLE_GENAI_USE_VERTEXAI=TRUE
GOOGLE_CLOUD_PROJECT={your-project-id}
GOOGLE_CLOUD_LOCATION=us-central1
STAGING_BUCKET=gs://purchasing-concierge-{your-project-id}
PIZZA_SELLER_AGENT_URL=https://pizza-agent-xxxxx.us-central1.run.app
BURGER_SELLER_AGENT_URL=https://burger-agent-xxxxx.us-central1.run.app
AGENT_ENGINE_RESOURCE_NAME=projects/xxxx/locations/us-central1/reasoningEngines/yyyy

Después, ejecuta el siguiente comando para implementar una app de Gradio

uv run purchasing_concierge_ui.py

Si la operación se realiza correctamente, se mostrará el siguiente resultado:

* Running on local URL:  http://0.0.0.0:8080
* To create a public link, set `share=True` in `launch()`.

Luego, presiona Ctrl + clic en la URL http://0.0.0.0:8080 en la terminal o haz clic en el botón de vista previa web para abrir la IU web.

Intenta tener una conversación como la siguiente :

Muéstrame el menú de hamburguesas y pizzas.
Quiero pedir 1 pizza de pollo a la barbacoa y 1 hamburguesa cajún picante

Y continúa la conversación hasta que finalices el pedido. Inspecciona cómo va la interacción y cuál es la llamada y la respuesta de la herramienta. La siguiente imagen es un ejemplo del resultado de la interacción.

Podemos ver que comunicarse con 2 agentes diferentes genera 2 comportamientos diferentes, y A2A puede manejar esto bien. El agente de venta de pizzas acepta directamente nuestra solicitud de agente de compra, mientras que el agente de hamburguesas necesita nuestra confirmación antes de proceder con nuestra solicitud y, después de que confirmamos, el agente puede confiar en la confirmación para el agente de hamburguesas.

Ahora, terminamos con los conceptos básicos de A2A y vemos cómo se implementa como arquitectura de cliente y servidor

8. Desafío

Ahora, ¿puedes preparar el archivo necesario e implementar la app de Gradio en Cloud Run por tu cuenta? ¡Es hora de aceptar el desafío!

9. Limpia

Sigue estos pasos para evitar que se apliquen cargos a tu cuenta de Google Cloud por los recursos que usaste en este codelab:

En la consola de Google Cloud, ve a la página Administrar recursos.
En la lista de proyectos, elige el proyecto que deseas borrar y haz clic en Borrar.
En el diálogo, escribe el ID del proyecto y, luego, haz clic en Cerrar para borrarlo.
Como alternativa, puedes ir a Cloud Run en la consola, seleccionar el servicio que acabas de implementar y borrarlo.