Use um agente Agent2Agent

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

Antes de começar

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

Operações compatíveis

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

Configuração

SDK Vertex AI para Python

Pode interagir com agentes A2A implementados no Agent Engine através do SDK Vertex AI para Python, usando a mesma sintaxe.

Para configurar a A2A com o Agent Engine, obtenha uma instância do agente implementado. Esta instância envolve os pontos finais A2A subjacentes, o que lhe permite chamar os pontos finais 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 Python A2A

Este método usa o SDK Python A2A oficial, que fornece uma biblioteca 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, peça o cartão do agente para criar uma instância de cliente. O A2AClient trata da descoberta e da comunicação por si.

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 pedidos Python

O protocolo A2A é criado com base em pontos finais HTTP padrão. Pode interagir com estes pontos finais através de qualquer cliente HTTP.

Obtenha o URL A2A do cartão do agente e defina os cabeçalhos do pedido.

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",
}

Recupere o cartão de agente

Tenha em atenção que o Agent Engine não publica o cartão de agente público. Para obter o cartão de agente autenticado:

SDK Vertex AI para Python 

response = await remote_agent.handle_authenticated_agent_card()

SDK Python A2A

response = await a2a_client.get_card()

Biblioteca de pedidos Python

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

Enviar mensagem

Para enviar uma mensagem:

SDK 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 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 pedidos 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))

Receba uma tarefa

Para obter uma tarefa e o respetivo estado

SDK Vertex AI para Python 

task_data = {
    "id": task_id,
}

response = await remote_agent.on_get_task(**task_data)

SDK Python A2A

from a2a.types import TaskQueryParams

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

Biblioteca de pedidos 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))

Cancele uma tarefa

Para cancelar uma tarefa:

SDK Vertex AI para Python 

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

SDK Python A2A

from a2a.types import TaskQueryParams

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

Biblioteca de pedidos 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))

O que se segue?