使用 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，该 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))

后续步骤