Utiliser un agent Agent2Agent

En plus des instructions générales pour utiliser un agent, cette page décrit les fonctionnalités spécifiques aux agents A2A.

Avant de commencer

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

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.

Configuration

SDK Vertex AI pour Python

Vous pouvez interagir avec les agents A2A déployés dans Agent Engine à l'aide du SDK Vertex AI pour Python, en utilisant la même syntaxe.

Pour configurer A2A avec Agent Engine, obtenez une instance de l'agent déployé. Cette instance encapsule les points de terminaison A2A sous-jacents, ce qui vous permet d'appeler les points de terminaison en tant que méthodes 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

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

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