Agent2Agent-Agent verwenden

Hinweise

In dieser Anleitung wird davon ausgegangen, dass Sie die Anleitung unter folgenden Links gelesen und befolgt haben:

Instanz eines Agenten abrufen

Wenn Sie eine A2aAgent abfragen möchten, müssen Sie zuerst eine neue Instanz erstellen oder eine vorhandene Instanz abrufen.

So rufen Sie die A2aAgent für eine bestimmte Ressourcen-ID ab:

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

Bei dieser Methode wird das offizielle A2A Python SDK verwendet, das eine Clientbibliothek für die Interaktion mit A2A-kompatiblen Agents 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 Agents ab, um eine Clientinstanz zu erstellen. Die 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 diesen Endpunkten über einen beliebigen HTTP-Client interagieren.

Rufen Sie die A2A-URL von der Agent-Karte 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:

  • eine agent.api_resource mit Informationen zum bereitgestellten Agent. Sie können auch agent.operation_schemas() aufrufen, um die Liste der Vorgänge zurückzugeben, die vom Agent unterstützt werden. Weitere Informationen finden Sie unter Unterstützte Vorgänge.
  • eine agent.api_client, die synchrone Dienstinteraktionen ermöglicht
  • ein agent.async_api_client, das asynchrone Dienstinteraktionen ermöglicht

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

Unterstützte Vorgänge

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

Agentenkarte abrufen

Die öffentliche Agent-Karte wird nicht von Agent Engine bereitgestellt. So rufen Sie die authentifizierte Agent-Karte 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

Aufgabe und Status abrufen

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