Agent2Agent-Agent verwenden

Hinweis

In dieser Anleitung wird davon ausgegangen, dass Sie die Anweisungen in den folgenden Dokumenten gelesen und ausgeführt haben:

Instanz eines Agent abrufen

Um einen A2aAgent abzufragen, müssen Sie zuerst eine neue Instanz erstellen oder eine vorhandene Instanz abrufen.

So rufen Sie den A2aAgent ab, der einer bestimmten Ressourcen-ID entspricht:

Vertex AI SDK für 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)

Dabei gilt:

A2A Python SDK

Diese Methode verwendet das offizielle A2A Python SDK, das eine Clientbibliothek für die Interaktion mit A2A-konformen Agenten bietet. Weitere Informationen finden Sie in der Dokumentation zum A2A Python SDK.

Installieren Sie zuerst das SDK:

pip install a2a-sdk>=0.3.4

Rufen Sie dann die Karte des Agent ab, um eine Clientinstanz zu erstellen. Der A2AClient übernimmt die Erkennung und Kommunikation für Sie.

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)

Python-Bibliothek „requests“

Das A2A-Protokoll basiert auf Standard-HTTP-Endpunkten. Sie können mit jedem HTTP-Client mit diesen Endpunkten interagieren.

Rufen Sie die A2A-URL von der Agentenkarte ab und definieren Sie die Anfrageheader.

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

Wenn Sie das Vertex AI SDK für Python verwenden, entspricht das remote_agent Objekt einer AgentEngine Klasse, die Folgendes enthält:

Im weiteren Verlauf dieses Abschnitts wird davon ausgegangen, dass Sie eine AgentEngine-Instanz mit dem Namen remote_agent haben.

Unterstützte Vorgänge

Ein A2A-Agent, der in Agent Engine gehostet wird, stellt eine Reihe von Vorgängen bereit, die direkt den API-Endpunkten des A2A-Protokolls entsprechen.

Agentenkarte abrufen

Agent Engine stellt die öffentliche Agentenkarte nicht bereit. So rufen Sie die authentifizierte Agentenkarte ab:

Vertex AI SDK für Python 

response = await remote_agent.handle_authenticated_agent_card()

A2A Python SDK

response = await a2a_client.get_card()

Python-Bibliothek „requests“

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

Nachricht senden

So senden Sie eine Nachricht:

Vertex AI SDK für 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)

A2A Python SDK

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)

Python-Bibliothek „requests“

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

Aufgabe abrufen

So rufen Sie eine Aufgabe und ihren Status ab:

Vertex AI SDK für Python 

task_data = {
    "id": task_id,
}

response = await remote_agent.on_get_task(**task_data)

A2A Python SDK

from a2a.types import TaskQueryParams

task_data ={
    "id":task_id,
}
response = await a2a_client.get_task(TaskQueryParams(**task_data))

Python-Bibliothek „requests“

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

Aufgabe abbrechen

So brechen Sie eine Aufgabe ab:

Vertex AI SDK für Python 

task_data = {
    "id": task_id,
}
response = await remote_agent.on_cancel_task(**task_data)

A2A Python SDK

from a2a.types import TaskQueryParams

task_data ={
    "id":task_id,
}
response = await a2a_client.cancel_task(TaskQueryParams(**task_data))

Python-Bibliothek „requests“

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

Nächste Schritte