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

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

1. 📖 Introducción

El protocolo Agent2Agent (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 LLM con datos y recursos. A2A intenta complementar el MCP, ya que A2A se enfoca en un problema diferente. Mientras que el MCP se enfoca en reducir la complejidad para conectar agentes con herramientas y datos, 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 se espera en este instructivo

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 el proyecto de Google Cloud
Configura el directorio de trabajo para el entorno de codificación
Implementa el agente de hamburguesas en Cloud Run
Implementa el agente de pizzas en Cloud Run
Implementa el asistente de compras en Agent Engine
Interactúa con el asistente de compras a través de la interfaz local

Descripción general de la arquitectura

Implementarás la siguiente arquitectura de servicio

Implementarás 2 servicios que actuarán como servidor A2A: el agente de hamburguesas ( respaldado por el framework de agentes de CrewAI) y el agente de pizza ( 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 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 agente 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 una cuenta de 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. 🚀 Preparación de la configuración de desarrollo del taller

Paso 1: Selecciona el proyecto activo en la consola de Cloud

En la página del selector de proyectos de la consola de Google Cloud, selecciona o crea un proyecto de Google Cloud (consulta la sección superior izquierda de la consola).

Haz clic en él y verás una lista de todos tus proyectos, como en este ejemplo:

El valor que se indica con el cuadro rojo es el ID DEL PROYECTO, y se usará en todo el instructivo.

Asegúrate de que la facturación esté habilitada para tu proyecto de Cloud. Para verificarlo, haz clic en el ícono de hamburguesa ☰ en la barra superior izquierda, que muestra el menú de navegación, y busca el menú Facturación.

Si ves el mensaje "La cuenta de facturación de prueba de Google Cloud Platform" está vinculada, tu proyecto está listo para usarse en este instructivo. Si no es así, vuelve al inicio de este instructivo y canjea la cuenta de facturación.

Paso 2: Familiarízate con Cloud Shell

Usarás Cloud Shell durante la mayor parte de los instructivos. 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, deberemos verificar si el shell ( o la terminal) ya se autenticó con nuestra cuenta.

gcloud auth list

Si ves tu Gmail personal como en el siguiente ejemplo de resultado, todo está bien.

Credentialed Accounts

ACTIVE: *
ACCOUNT: alvinprayuda@gmail.com

To set the active account, run:
    $ gcloud config set account `ACCOUNT`

Si no es así, intenta actualizar el navegador y asegúrate de hacer clic en Autorizar cuando se te solicite ( es posible que se interrumpa debido a un problema de conexión).

A continuación, también debemos verificar si la shell ya está configurada con el ID DEL PROYECTO correcto que tienes. Si ves que hay un valor dentro de ( ) antes del ícono de $ en la terminal ( en la captura de pantalla a continuación, el valor es "a2a-agent-engine"), este valor muestra el proyecto configurado para tu sesión de shell activa.

Si el valor que se muestra ya es correcto, puedes omitir el siguiente comando. Sin embargo, si no es correcto o falta, ejecuta el siguiente comando:

gcloud config set project <YOUR_PROJECT_ID>

Luego, clona 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

Paso 3: Familiarízate con el 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 .

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, luego, el directorio purchasing-concierge-a2a. Por último, haz clic en el botón Aceptar. Esto convertirá el directorio elegido en 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í:

Ahora, abre la terminal del editor. Para ello, haz clic en Terminal -> New Terminal en la barra de menú o usa Ctrl + Mayúsculas + C. Se abrirá una ventana de terminal en la parte inferior del navegador.

Tu terminal activa actual debe estar dentro del directorio de trabajo purchasing-concierge-a2a. Usaremos Python 3.12 en este codelab y uv python project manager para simplificar la necesidad de crear y administrar la versión de Python y el entorno virtual. Este paquete uv ya está preinstalado en Cloud Shell.

Ejecuta este comando para instalar las dependencias requeridas en el entorno virtual del directorio .venv.

uv sync --frozen

Consulta el archivo pyproject.toml para ver las dependencias declaradas para este instructivo, que son a2a-sdk, google-adk, and gradio.

Ahora, deberemos habilitar las APIs requeridas con el siguiente comando. Este proceso podría tardar un poco.

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.

3. 🚀 Implementación de agentes de vendedores remotos del 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.

4. 🚀 Implementación del agente de Burger Seller: servidor A2A

El código fuente del agente de hamburguesas se encuentra en el directorio remote_seller_agents/burger_agent.

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. 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. 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.

Abre una nueva pestaña del navegador y ve 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 la tarjeta del agente.

Esta es la información de la tarjeta del agente de hamburguesas a la que se debe poder acceder para fines de descubrimiento.

Ten en cuenta 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.

Debemos 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.

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 url ya estará configurado correctamente.

5. 🚀 Implementa el agente de vendedor de pizzas: servidor A2A

Del mismo modo, el código fuente del agente de pizza se encuentra en el directorio remote_seller_agents/pizza_agent.

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 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.

Cómo actualizar 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.

6. 🚀 Implementación del asistente 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 compra se encuentra en el directorio purchasing_concierge. La inicialización del agente se puede inspeccionar en la secuencia de comandos purchasing_concierge/purchasing_agent.py.

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, visita 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. Lo implementaremos en el motor del agente, y el código de implementación se encuentra dentro del script deploy_to_agent_engine.py.

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. Se mostrará el nombre del recurso de Agent Engine como "projects/xxxx/locations/us-central1/reasoningEngines/yyyy".

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.

También puedes inspeccionar que el nombre del recurso de Agent Engine se muestre allí. Luego, podemos usar este nombre de recurso para probarlo.

Luego, actualiza el AGENT_ENGINE_RESOURCE_NAME en el archivo .env con este valor. 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

Prueba del 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. Ejecuta la siguiente secuencia de comandos test_agent_engine.sh:

bash test_agent_engine.sh

Puedes inspeccionar el script y ver que intentamos preguntarle "Enumera el menú de hamburguesas disponible, por favor" al agente.

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.

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. Ejecuta el siguiente comando para implementar una app de Gradio. Para ejecutar esta app, ya debes haber completado el archivo .env correctamente.

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 que terminamos con los conceptos básicos de A2A, veamos cómo se implementa como arquitectura de cliente y servidor.

8. 💡 [Explicación del código] Concepto e implementación del servidor A2A

La inicialización del agente de vendedor remoto se puede inspeccionar en la secuencia de comandos remote_seller_agents/*/agent.py. Este es el fragmento de código de los agentes vendedores.

Agente de hamburguesas

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

...

Agente de pizza

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

...

Como puedes ver, estos 2 agentes se compilan con frameworks completamente diferentes ( CrewAI y Langgraph) en comparación con el agente del cliente ( ADK). Con A2A, esto no es un problema. No es necesario que compartan su código interno para comunicarse entre sí, no importa qué frameworks se usen, qué idioma se utilice ni dónde se implementen.

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 pueda comunicarse 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 puede 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 una 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) .

9. 💡 [Explicación del código] Implementación de Agent Engine

Este es el fragmento de código del agente de asistente de compra en purchasing_concierge/purchasing_agent.py:

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

...

Este agente se compila con el ADK y se implementa 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.

En este instructivo, realizamos la implementación con la secuencia de comandos deploy_to_agent_engine.py, cuyo contenido se muestra a continuación.

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 del ADK en Agent Engine. Primero, necesitamos 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 del directorio del agente en extra_packages ( también puedes proporcionar otros directorios y archivos si es necesario aquí) y proporcionando las variables de entorno necesarias.

10. 💡 [Explicación del código] Concepto e implementación 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.). 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, lo 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 Mensaje 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.

11. 🎯 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!

12. 🧹 Limpieza

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 y Agent Engine en la consola, seleccionar el servicio que acabas de implementar y borrarlo.