Usar um agente Agent2Agent

Além das instruções gerais para usar um agente, esta página descreve recursos específicos dos agentes A2A.

Antes de começar

Este tutorial pressupõe que você leu e seguiu as instruções em:

Operações suportadas

Um agente A2A hospedado no Agent Engine expõe um conjunto de operações que correspondem diretamente aos endpoints da API do protocolo A2A.

Configuração

SDK da Vertex AI para Python

É possível interagir com agentes A2A implantados no Agent Engine usando o SDK da Vertex AI para Python, com a mesma sintaxe.

Para configurar o A2A com o Agent Engine, receba uma instância do agente implantado. Essa instância encapsula os endpoints A2A subjacentes, permitindo que você os chame como métodos 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 do Python A2A

Esse método usa o SDK oficial do Python A2A, que fornece uma biblioteca de cliente para interagir com agentes compatíveis com A2A. Para mais informações, consulte a documentação do SDK Python A2A.

Primeiro, instale o SDK:

pip install a2a-sdk>=0.3.4

Em seguida, extraia o card do agente para criar uma instância de cliente. O A2AClient processa a descoberta e a comunicação para você.

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 solicitações do Python

O protocolo A2A é baseado em endpoints HTTP padrão. É possível interagir com esses endpoints usando qualquer cliente HTTP.

Recupere o URL do A2A do card do agente e defina os cabeçalhos da solicitação.

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 o card do agente

O Agent Engine não veicula o card do agente público. Para recuperar o card do agente autenticado:

SDK da Vertex AI para Python 

response = await remote_agent.handle_authenticated_agent_card()

SDK do Python A2A

response = await a2a_client.get_card()

Biblioteca de solicitações do Python

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

Enviar uma mensagem

Para enviar uma mensagem, siga estas etapas:

SDK da 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 do 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 solicitações do 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))

Acessar uma tarefa

Para receber uma tarefa e o status dela

SDK da Vertex AI para Python 

task_data = {
    "id": task_id,
}

response = await remote_agent.on_get_task(**task_data)

SDK do Python A2A

from a2a.types import TaskQueryParams

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

Biblioteca de solicitações do 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 uma tarefa

Para cancelar uma tarefa:

SDK da Vertex AI para Python 

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

SDK do Python A2A

from a2a.types import TaskQueryParams

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

Biblioteca de solicitações do 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))

A seguir