Usa un agente Agent2Agent (A2A)

Antes de comenzar

En este instructivo, se supone que leíste y seguiste las instrucciones que se indican en los siguientes documentos:

Obtén una instancia de un agente

Para consultar un A2aAgent, primero debes crear una instancia nueva o obtener una instancia existente.

Para obtener el A2aAgent que corresponde a un ID de recurso específico, haz lo siguiente:

SDK de Agent Platform 

import agentplatform
from agentplatform import types

PROJECT_ID = "PROJECT_ID"
LOCATION = "LOCATION"
RESOURCE_ID = "RESOURCE_ID"
RESOURCE_NAME = f"projects/{PROJECT_ID}/locations/{LOCATION}/reasoningEngines/{RESOURCE_ID}"

client = agentplatform.Client(
    project=PROJECT_ID,
    location=LOCATION,
    http_options={"api_version": "v1beta1"},
)

remote_agent = client.runtimes.get(name=RESOURCE_NAME)

print(remote_agent)

donde

SDK de Python para A2A

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

Primero, instala el SDK:

pip install a2a-sdk>=0.3.4

Luego, obtén la tarjeta del agente para crear una instancia del cliente. A2AClient se encarga del descubrimiento y la comunicación por ti.

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 extremos HTTP estándar. Puedes interactuar con estos extremos usando 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",
}

Cuando se usa el SDK de Agent Platform, el objeto remote_agent corresponde a una clase AgentRuntime que contiene lo siguiente:

En el resto de esta sección, se supone que tienes una instancia de AgentRuntime, llamada remote_agent.

Operaciones admitidas

Un agente A2A alojado en Agent Runtime expone un conjunto de operaciones que corresponden directamente a los extremos de la API del protocolo A2A.

Recupera la tarjeta de agente

Ten en cuenta que Agent Runtime no muestra la tarjeta del agente público. Para recuperar la tarjeta de agente autenticada, haz lo siguiente:

SDK de Agent Platform 

response = await remote_agent.handle_authenticated_agent_card()

SDK de Python para 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))

Envía un mensaje

Para enviar un mensaje, sigue estos pasos:

SDK de Agent Platform 

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)

Puedes anular el tiempo de espera que se usa para on_message_send configurando el campo timeout de HttpOptions cuando creas agentplatform.Client.

SDK de Python para 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

Cómo obtener una tarea y su estado

SDK de Agent Platform 

task_data = {
    "id": task_id,
}

response = await remote_agent.on_get_task(**task_data)

SDK de Python para 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))

Cómo cancelar una tarea

Para cancelar una tarea, sigue estos pasos:

SDK de Agent Platform 

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

SDK de Python para 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))

¿Qué sigue?