Agent2Agent-Agent verwenden

Zusätzlich zur allgemeinen Anleitung zur Verwendung eines Agenten werden auf dieser Seite Funktionen beschrieben, die speziell für A2A-Agenten gelten.

Hinweise

In dieser Anleitung wird davon ausgegangen, dass Sie die Anleitung unter folgenden Links gelesen und befolgt 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.

Einrichtung

Vertex AI SDK für Python

Sie können mit A2A-Agents, die in Agent Engine bereitgestellt werden, über das Vertex AI SDK für Python interagieren. Dabei wird dieselbe Syntax verwendet.

Um A2A mit Agent Engine einzurichten, müssen Sie eine Instanz des bereitgestellten Agents abrufen. Diese Instanz umschließt die zugrunde liegenden A2A-Endpunkte, sodass Sie die Endpunkte als Python-Methoden aufrufen können.

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

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

Agent-Karte 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