Utilizzare un agente Agent2Agent

Prima di iniziare

Questo tutorial presuppone che tu abbia letto e seguito le istruzioni riportate in:

Ottenere un'istanza di un agente

Per eseguire query su un A2aAgent, devi prima creare una nuova istanza o ottenere un'istanza esistente.

Per ottenere il A2aAgent corrispondente a un ID risorsa specifico:

SDK Vertex AI Python 

import vertexai
from google.genai import types

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

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

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

print(remote_agent)

dove

SDK Python A2A

Questo metodo utilizza l'SDK Python A2A ufficiale, che fornisce una libreria client per interagire con gli agenti conformi ad A2A. Per ulteriori informazioni, consulta la documentazione dell'SDK Python A2A.

Innanzitutto, installa l'SDK:

pip install a2a-sdk>=0.3.4

Quindi, recupera la scheda dell'agente per creare un'istanza client. A2AClient gestisce la rilevabilità e la comunicazione per te.

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)

Libreria delle richieste Python

Il protocollo A2A è basato su endpoint HTTP standard. Puoi interagire con questi endpoint utilizzando qualsiasi client HTTP.

Recupera l'URL A2A dalla scheda dell'agente e definisci le intestazioni delle richieste.

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

Quando utilizzi l'SDK Vertex AI per Python, l'oggetto remote_agent corrisponde a una AgentEngine classe che contiene quanto segue:

Il resto di questa sezione presuppone che tu disponga di un'istanza AgentEngine, denominata remote_agent.

Operazioni supportate

Un agente A2A ospitato su Agent Engine espone un insieme di operazioni che corrispondono direttamente agli endpoint API del protocollo A2A.

Recuperare la scheda dell'agente

Tieni presente che Agent Engine non pubblica la scheda dell'agente pubblico. Per recuperare la scheda dell'agente autenticato:

SDK Vertex AI Python 

response = await remote_agent.handle_authenticated_agent_card()

SDK Python A2A

response = await a2a_client.get_card()

Libreria delle richieste Python

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

Inviare un messaggio

Per inviare un messaggio:

SDK Vertex AI 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)

Puoi eseguire l'override del timeout utilizzato per on_message_send impostando il campo timeout di HttpOptions durante la creazione di vertexai.Client.

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)

Libreria delle richieste 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))

Recuperare un'attività

Per recuperare un'attività e il relativo stato

SDK Vertex AI 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))

Libreria delle richieste 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))

Annullare un'attività

Per annullare un'attività:

SDK Vertex AI 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))

Libreria delle richieste 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))

Passaggi successivi