Utiliser un agent Agent2Agent (A2A)

Avant de commencer

Ce tutoriel suppose que vous avez lu et suivi les instructions de :

Obtenir une instance d'un agent

Pour interroger un A2aAgent, vous devez d'abord créer une instance ou en obtenir une existante.

Pour obtenir le A2aAgent qui correspond à un ID de ressource spécifique :

SDK Agent Platform 

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)

Où :

SDK Python A2A

Cette méthode utilise le SDK Python A2A officiel, qui fournit une bibliothèque cliente pour interagir avec les agents compatibles avec A2A. Pour en savoir plus, consultez la documentation du SDK Python A2A.

Commencez par installer le SDK :

pip install a2a-sdk>=0.3.4

Ensuite, obtenez la carte de l'agent pour créer une instance de client. A2AClient gère la découverte et la communication pour vous.

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)

Bibliothèque de requêtes Python

Le protocole A2A est basé sur des points de terminaison HTTP standards. Vous pouvez interagir avec ces points de terminaison à l'aide de n'importe quel client HTTP.

Récupérez l'URL A2A à partir de la carte de l'agent et définissez les en-têtes de requête.

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

Lorsque vous utilisez le SDK Agent Platform, l'objet remote_agent correspond à une AgentEngine classe qui contient les éléments suivants :

Le reste de cette section suppose que vous disposez d'une instance AgentEngine, nommée remote_agent.

Opérations prises en charge

Un agent A2A hébergé sur Agent Runtime expose un ensemble d'opérations qui correspondent directement aux points de terminaison d'API du protocole A2A.

Récupérer la carte de l'agent

Notez qu'Agent Runtime ne fournit pas la carte d'agent public. Pour récupérer la carte d'agent authentifié :

SDK Agent Platform 

response = await remote_agent.handle_authenticated_agent_card()

SDK Python A2A

response = await a2a_client.get_card()

Bibliothèque de requêtes Python

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

Envoyer un message

Pour envoyer un message, procédez comme suit :

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

Vous pouvez remplacer le délai avant expiration utilisé pour on_message_send en définissant le champ timeout de HttpOptions lors de la création de 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)

Bibliothèque de requêtes 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))

Obtenir une tâche

Pour obtenir une tâche et son état :

SDK Agent Platform 

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

Bibliothèque de requêtes 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))

Annuler une tâche

Pour annuler une tâche :

SDK Agent Platform 

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

Bibliothèque de requêtes 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))

Étape suivante