Usar un agente Agent2Agent

Además de las instrucciones generales para usar un agente, en esta página se describen las funciones específicas de los agentes de A2A.

Antes de empezar

En este tutorial se da por hecho que has leído y seguido las instrucciones de los siguientes artículos:

Operaciones admitidas

Un agente A2A alojado en Agent Engine expone un conjunto de operaciones que se corresponden directamente con los endpoints de la API del protocolo A2A.

Configuración

SDK de Vertex AI para Python

Puedes interactuar con los agentes de A2A desplegados en Agent Engine mediante el SDK de Vertex AI para Python, con la misma sintaxis.

Para configurar A2A con Agent Engine, obtén una instancia del agente implementado. Esta instancia encapsula los endpoints de A2A subyacentes, lo que te permite llamar a los endpoints como métodos de Python.

import vertexai
from google.genai import types

# Replace with your actual values
PROJECT_ID = "your-project-id"
LOCATION = "your-location"
REASONING_ENGINE_ID = "your-reasoning-engine-id"
AGENT_ENGINE_RESOURCE = f"projects/{PROJECT_ID}/locations/{LOCATION}/reasoningEngines/{REASONING_ENGINE_ID}"

client = vertexai.Client(
    project=PROJECT_ID,
    location=LOCATION,
    http_options=types.HttpOptions(
        api_version="v1beta1")
)

remote_agent = client.agent_engines.get(
    name=AGENT_ENGINE_RESOURCE,
)

SDK de Python A2A

Este método usa el SDK oficial de Python de A2A, que proporciona una biblioteca de cliente para interactuar con agentes compatibles con A2A. Para obtener más información, consulta la documentación del SDK de Python de A2A.

Primero, instala el SDK:

pip install a2a-sdk>=0.3.4

A continuación, obtén la tarjeta del agente para crear una instancia de cliente. A2AClient se encarga de la detección y la comunicación.

from google.auth import default
from google.auth.transport.requests import Request
from a2a.client import ClientConfig, ClientFactory
from a2a.types import TransportProtocol
import httpx

# We assume 'agent_card' is an existing AgentCard object.

# Fetch credentials for authentication for demo purpose. Use your own auth
credentials, _ = default(scopes=['https://www.googleapis.com/auth/cloud-platform'])
credentials.refresh(Request())

# Create the client by chaining the factory and config initialization.
factory = ClientFactory(
    ClientConfig(
        supported_transports=[TransportProtocol.http_json], # only support http_json
        use_client_preference=True,
        httpx_client=httpx.AsyncClient(
            headers={
                "Authorization": f"Bearer {credentials.token}",
                "Content-Type": "application/json",
            }
        ),
    )
)
a2a_client = factory.create(agent_card)

Biblioteca de solicitudes de Python

El protocolo A2A se basa en endpoints HTTP estándar. Puedes interactuar con estos endpoints mediante cualquier cliente HTTP.

Recupera la URL de A2A de la tarjeta del agente y define los encabezados de la solicitud.

from google.auth import default
from google.auth.transport.requests import Request

# We assume 'agent_card' is an existing object
a2a_url = agent_card.url

# Get an authentication token for demonstration purposes. Use your own authentication mechanism.
credentials, _ = default(scopes=['https://www.googleapis.com/auth/cloud-platform'])
credentials.refresh(Request())

headers = {
    "Authorization": f"Bearer {credentials.token}",
    "Content-Type": "application/json",
}

Recuperar la tarjeta del agente

Ten en cuenta que Agent Engine no sirve la tarjeta de agente público. Para obtener la tarjeta del agente autenticado, sigue estos pasos:

SDK de Vertex AI para Python 

response = await remote_agent.handle_authenticated_agent_card()

SDK de Python A2A

response = await a2a_client.get_card()

Biblioteca de solicitudes de Python

card_endpoint = f"{a2a_url}/v1/card"
response = httpx.get(card_endpoint, headers=headers)
print(json.dumps(response.json(), indent=4))

Enviar un mensaje

Sigue estos pasos para enviar un mensaje:

SDK de Vertex AI para Python 

message_data = {
  "messageId": "remote-agent-message-id",
  "role": "user",
  "parts": [{"kind": "text", "text": "What is the exchange rate from USD to EUR today?"}],
}

response = await remote_agent.on_message_send(**message_data)

SDK de Python A2A

from a2a.types import Message, Part, TextPart
import pprint

message = Message(
    message_id="remote-agent-message-id",
    role="user",
    parts=[Part(root=TextPart(text="What's the currency rate of USD and EUR"))],
)

response_iterator = a2a_client.send_message(message)

async for chunk in response_iterator:
    pprint.pp(chunk)

Biblioteca de solicitudes de Python

import httpx
import json

endpoint = f"{a2a_url}/v1/message:send"

payload = {
    "message": {
        "messageId": "remote-agent-message-id",
        "role": "1",
        "content": [{"text": "What is the exchange rate from USD to EUR today?"}],
    },
    "metadata": {"source": "python_script"},
}

response = httpx.post(endpoint, json=payload, headers=headers)
print(json.dumps(response.json(), indent=4))

Obtener una tarea

Para obtener una tarea y su estado

SDK de Vertex AI para Python 

task_data = {
    "id": task_id,
}

response = await remote_agent.on_get_task(**task_data)

SDK de Python A2A

from a2a.types import TaskQueryParams

task_data ={
    "id":task_id,
}
response = await a2a_client.get_task(TaskQueryParams(**task_data))

Biblioteca de solicitudes de Python

task_end_point = f"{a2a_url}/v1/tasks/{task_id}"
response = httpx.get(task_end_point, headers=headers)
print(json.dumps(response.json(), indent=4))

Cancelar una tarea

Para cancelar una tarea, sigue estos pasos:

SDK de Vertex AI para Python 

task_data = {
    "id": task_id,
}
response = await remote_agent.on_cancel_task(**task_data)

SDK de Python A2A

from a2a.types import TaskQueryParams

task_data ={
    "id":task_id,
}
response = await a2a_client.cancel_task(TaskQueryParams(**task_data))

Biblioteca de solicitudes de Python

task_end_point = f"{a2a_url}/v1/tasks/{task_id}:cancel"
response = httpx.post(task_end_point, headers=headers)
print(json.dumps(response.json(), indent=4))

Siguientes pasos