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 obtenir une instance existante.

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

SDK Agent Platform 

import agentplatform
from agentplatform import types

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

client = agentplatform.Client(
    project=PROJECT_ID,
    location=LOCATION,
    http_options={"api_version": "v1beta1"},
)

remote_agent = client.runtimes.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 conformes à 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, récupérez la fiche de l'agent pour créer une instance 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 s'appuie 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 fiche 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 classe AgentRuntime qui contient les éléments suivants :

Le reste de cette section suppose que vous disposez d'une instance AgentRuntime 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 de l'API du protocole A2A.

Récupérer la carte de l'agent

Notez que l'Agent Runtime ne diffuse pas la fiche publique de l'agent. Pour récupérer la fiche de l'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 d'inactivité utilisé pour on_message_send en définissant le champ timeout de HttpOptions lors de la création de agentplatform.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))

Étapes suivantes