Utiliser un agent Agent2Agent

Avant de commencer

Pour suivre ce tutoriel, vous devez avoir lu et suivi les instructions des ressources suivantes :

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 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 A2A pour Python

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

Commencez par installer le SDK :

pip install a2a-sdk>=0.3.4

Ensuite, obtenez la fiche de l'agent pour créer une instance cliente. 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 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 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 Engine expose un ensemble d'opérations qui correspondent directement aux points de terminaison d'API du protocole A2A.

Récupérer la fiche de l'agent

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

SDK Vertex AI pour Python 

response = await remote_agent.handle_authenticated_agent_card()

SDK A2A pour Python

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 A2A pour Python

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 A2A pour Python

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 A2A pour Python

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