Utilizzare un agente Agent2Agent

Oltre alle istruzioni generali per utilizzare un agente, questa pagina descrive le funzionalità specifiche degli agenti A2A.

Prima di iniziare

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

Operazioni supportate

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

Configurazione

SDK Vertex AI per Python

Puoi interagire con gli agenti A2A di cui è stato eseguito il deployment in Agent Engine utilizzando l'SDK Vertex AI per Python, con la stessa sintassi.

Per configurare A2A con Agent Engine, recupera un'istanza dell'agente di cui è stato eseguito il deployment. Questa istanza esegue il wrapping degli endpoint A2A sottostanti, consentendoti di chiamare gli endpoint come metodi 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

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

Poi, ottieni la scheda dell'agente per creare un'istanza client. A2AClient gestisce l'individuazione 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 si basa 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 della richiesta.

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

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 per 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))

Invia un messaggio

Per inviare un messaggio:

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

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 ottenere un'attività e il relativo stato

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