使用 Agent2Agent 代理程式

除了使用代理程式的一般操作說明外，本頁面還會說明 A2A 代理程式專屬的功能。

事前準備

本教學課程假設您已詳閱並按照下列教學課程的指示操作：

支援的作業

Agent Engine 託管的 A2A 代理會公開一組作業，直接對應至 A2A 通訊協定的 API 端點。

設定

Python 適用的 Vertex AI SDK

您可以使用 Vertex AI SDK for Python，以相同語法與部署至 Agent Engine 的 A2A 代理程式互動。

如要使用 Agent Engine 設定 A2A，請取得已部署代理程式的執行個體。這個執行個體會包裝基礎 A2A 端點，讓您將端點當做 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,
)

A2A Python SDK

這個方法會使用官方 A2A Python SDK，提供用戶端程式庫與符合 A2A 規範的代理程式互動。詳情請參閱 A2A Python SDK 說明文件。

首先，請安裝 SDK：

pip install a2a-sdk>=0.3.4

接著，取得代理程式的卡片，建立用戶端例項。A2AClient 會為您處理探索和通訊事宜。

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 requests 程式庫

A2A 通訊協定是以標準 HTTP 端點為基礎建構而成。您可以使用任何 HTTP 用戶端與這些端點互動。

從代理程式資訊卡擷取 A2A 網址，並定義要求標頭。

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 Engine 不會提供公開的代理程式卡片。如要擷取已通過驗證的代理程式資訊卡：

Python 適用的 Vertex AI SDK 

response = await remote_agent.handle_authenticated_agent_card()

A2A Python SDK

response = await a2a_client.get_card()

Python requests 程式庫

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

傳送訊息

如要傳送訊息，請按照下列步驟操作：

Python 適用的 Vertex AI SDK 

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

取得工作

如要取得工作及其狀態

Python 適用的 Vertex AI SDK 

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

取消工作

如要取消工作：

Python 適用的 Vertex AI SDK 

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

後續步驟