Utiliser un agent Agent2Agent

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

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 Vertex AI pour Python, l'objet remote_agent correspond à une classe AgentEngine qui contient les éléments suivants :

  • un agent.api_resource contenant des informations sur l'agent déployé. Vous pouvez également appeler agent.operation_schemas() pour renvoyer la liste des opérations compatibles avec l'agent. Pour en savoir plus, consultez Opérations compatibles.
  • un agent.api_client qui permet des interactions de service synchrones.
  • un agent.async_api_client qui permet des interactions de service asynchrones.

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

Opérations compatibles

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

Récupérer la fiche de l'agent

Notez qu'Agent Engine ne diffuse pas la fiche d'agent public. Pour récupérer la fiche de l'agent authentifié :

SDK Vertex AI pour Python 

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

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

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

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