使用代理的注册 A2A 端点调用代理

Agent Registry 中的 Agent2Agent (A2A) 代理会宣传包含端点网址和协议绑定（例如 HTTP_JSON）的协议接口。您可以在注册表中发现代理的网址，以便从自定义编排器或客户端调用其 A2A 方法。

概览

规范 详细信息
Discovery API agentregistry.googleapis.com (v1)
调用代理主机 LOCATION-discoveryengine.googleapis.com
协议绑定 HTTP_JSON
网址项目标识符 Google Cloud 项目编号（而非项目 ID）
支持的 A2A 方法 GET /v1/card、POST /v1/message:send、POST /v1/message:stream
必需的消息架构 message.role = "ROLE_USER"、content[].text、唯一 messageId
必需的 IAM 权限 roles/agentregistry.viewer（发现）和 discoveryengine.assistants.assist（调用）

准备工作

  1. 在您的 Google Cloud 项目中启用 Agent Registry API (agentregistry.googleapis.com) 和 Discovery Engine API (discoveryengine.googleapis.com)。
  2. 如果代理不是直接在 Gemini Enterprise 应用中创建的，请从 Agent Registry 导入代理，并授予最终用户访问权限。如需查看相关说明，请参阅从 Agent Registry 导入 A2A 代理。
  3. 向调用方主账号授予适当的 IAM 权限：
    • 读取注册表：Agent Registry Viewer (roles/agentregistry.viewer)。
    • 调用代理：Discovery Engine Editor (roles/discoveryengine.editor) 或包含 discoveryengine.assistants.assist 的自定义角色。
  4. 如果您使用应用默认凭据 (ADC) 进行身份验证，请将客户端配置为发送配额项目标头：-H "X-Goog-User-Project: PROJECT_ID"。
  5. （可选）如果您计划将远程代理封装为程序化子代理，请安装智能体开发套件 (ADK) 库：pip install "google-adk[a2a]>=1.29.0"。

第 1 步：发现代理及其 A2A 端点

如需调用 A2A 智能体，请先在 Agent Registry 中发现其宣传的 url。列出注册表位置（例如 us 或 eu，作为全局 agentregistry.googleapis.com 主机上的路径参数传递）中的代理：

curl -X GET \
-H "Authorization: Bearer $(gcloud auth print-access-token)" \
-H "Content-Type: application/json" \
"https://agentregistry.googleapis.com/v1/projects/PROJECT_ID/locations/LOCATION/agents?pageSize=100"

您还可以使用 Google Cloud CLI 按显示名称前缀搜索代理：

gcloud agent-registry agents search \
  --project=PROJECT_ID \
  --location=LOCATION \
  --search-string="displayName:My_Agent_*"

在返回的代理资源中，检查 protocols 数组。找到 type 等于 A2A_AGENT 且 interfaces[].protocolBinding 等于 HTTP_JSON 的条目。提取相应的 url：

{
  "name": "projects/PROJECT_ID/locations/LOCATION/agents/AGENT_RESOURCE_ID",
  "displayName": "My Agent",
  "protocols": [
    {
      "type": "A2A_AGENT",
      "protocolVersion": "0.3.0",
      "interfaces": [
        {
          "url": "https://LOCATION-discoveryengine.googleapis.com/v1/projects/PROJECT_NUMBER/locations/LOCATION/collections/default_collection/engines/ENGINE_ID/assistants/default_assistant/agents/AGENT_ID/a2a",
          "protocolBinding": "HTTP_JSON"
        }
      ]
    }
  ]
}

如需查看完整的代理资源架构，请参阅 projects.locations.agents REST API 参考文档。

第 2 步：提取代理卡片

代理卡片提供元数据，用于描述代理的身份、说明以及输入/输出功能。如需检索卡片，请向附加到代理的 A2A 端点网址的 /v1/card 路径发送 HTTP GET 请求：

curl -X GET \
-H "Authorization: Bearer $(gcloud auth print-access-token)" \
"A2A_ENDPOINT_URL/v1/card"

响应载荷示例：

{
  "name": "My Agent",
  "description": "What the agent does.",
  "url": "A2A_ENDPOINT_URL",
  "capabilities": {},
  "defaultInputModes": ["text"],
  "defaultOutputModes": ["text"],
  "preferredTransport": "HTTP+JSON"
}

第 3 步：发送消息

如需向代理发送用户查询，请向 /v1/message:send 发出 POST 请求。请求正文必须符合 A2A 消息架构，要求 role 设置为 ROLE_USER、包含文本部分的 content 数组以及唯一生成的 messageId：

curl -X POST \
-H "Authorization: Bearer $(gcloud auth print-access-token)" \
-H "Content-Type: application/json" \
"A2A_ENDPOINT_URL/v1/message:send" \
  -d '{
    "message": {
      "role": "ROLE_USER",
      "content": [
        {
          "text": "What can you help me with?"
        }
      ],
      "messageId": "UNIQUE_UUID_STRING"
    }
  }'

在响应载荷中，代理的回复会在 message 对象内返回：

{
  "message": {
    "contextId": "projects/PROJECT_NUMBER/locations/LOCATION/collections/default_collection/engines/ENGINE_ID/sessions/SESSION_ID",
    "role": "ROLE_AGENT",
    "content": [
      {
        "text": "I am an AI assistant..."
      }
    ]
  }
}

连接 content[].text 内的文本字符串以显示完整响应。如需在同一会话中继续对话，请保存返回的 contextId 字符串，并在下一次请求中将其作为 message.contextId 提供。

如需查看完整消息载荷架构，请参阅 A2A message:send REST API 参考文档。

以增量方式流式传输响应

对于流式输出，请向 /v1/message:stream 发送具有相同消息正文的 POST 请求：

curl -N -X POST \
-H "Authorization: Bearer $(gcloud auth print-access-token)" \
-H "Content-Type: application/json" \
"A2A_ENDPOINT_URL/v1/message:stream" \
  -d '{
    "message": {
      "role": "ROLE_USER",
      "content": [
        {
          "text": "Say hello."
        }
      ],
      "messageId": "UNIQUE_UUID_STRING"
    }
  }'

该端点通过 HTTP 返回流式传输的块对象的 JSON 数组。按顺序附加 content[].text 片段。流式传输的块还包含 metadata.sessionInfo 和 metadata.assistToken。

如需查看流式传输载荷规范，请参阅 A2A message:stream REST API 参考文档。

使用 Python 调用 A2A 端点

此 Python 脚本解析 Agent Registry 中的 A2A 端点，并使用原始 HTTP 请求发送消息：

# Install dependencies: pip install google-auth requests
import uuid
import google.auth
from google.auth.transport.requests import AuthorizedSession

# TODO(developer): Replace placeholder values with your project ID and location.
project_id = "PROJECT_ID"
location = "LOCATION"          # Registry location (for example: "us" or "eu")
target_display_name = "My Agent"
query_text = "What can you help me with?"

# Initialize credentials and authorized session
creds, _ = google.auth.default(
    scopes=["https://www.googleapis.com/auth/cloud-platform"]
)
session = AuthorizedSession(creds)

# Step 1: Resolve the A2A endpoint URL from the Agent Registry
registry_url = (
    f"https://agentregistry.googleapis.com/v1/"
    f"projects/{project_id}/locations/{location}/agents"
)
response = session.get(registry_url)
response.raise_for_status()
agents = response.json().get("agents", [])

def get_a2a_url(agent_resource):
    for proto in agent_resource.get("protocols") or []:
        if proto.get("type") == "A2A_AGENT":
            for iface in proto.get("interfaces", []):
                if iface.get("protocolBinding") == "HTTP_JSON":
                    return iface.get("url")
    return None

target_agent = next(
    (a for a in agents if a.get("displayName") == target_display_name),
    None
)
if not target_agent:
    raise SystemExit(f"Agent '{target_display_name}' not found in registry.")

endpoint_url = get_a2a_url(target_agent)
if not endpoint_url:
    raise SystemExit("Target agent does not publish an HTTP_JSON A2A endpoint.")

# Step 2: Fetch and verify the agent card
card_resp = session.get(f"{endpoint_url}/v1/card")
card_resp.raise_for_status()
card = card_resp.json()
print("Resolved Agent:", card.get("name"))

# Step 3: Send an A2A message
body = {
    "message": {
        "role": "ROLE_USER",
        "content": [{"text": query_text}],
        "messageId": str(uuid.uuid4()),
    }
}
send_resp = session.post(f"{endpoint_url}/v1/message:send", json=body)
send_resp.raise_for_status()

reply_message = send_resp.json().get("message", {})
full_reply_text = "".join(
    part.get("text", "") for part in reply_message.get("content", [])
)
print("Agent Reply:", full_reply_text)

使用 ADK 简化编排

智能体开发套件 (ADK) 会自动解析注册表端点，并将远程 A2A 代理封装为子代理：

from google.adk.integrations.agent_registry import AgentRegistry

# Initialize registry client
registry = AgentRegistry(project_id="PROJECT_ID", location="LOCATION")

# Resolve remote A2A agent directly by resource name
remote_agent = registry.get_remote_a2a_agent(
    agent_name="agents/AGENT_RESOURCE_ID"
)

补充说明

A2A 端点具有以下行为：

  • 严格的路径命名：对于 HTTP+JSON 绑定，仅支持 GET {url}/v1/card、POST {url}/v1/message:send 和 POST {url}/v1/message:stream。
  • 严格的架构验证：将纯 "user" 作为角色传递会返回 HTTP 400 Bad Request 错误。您必须传递枚举字符串 "ROLE_USER"。同样，消息文本必须位于 content 数组内，而不是 parts，并且严格要求提供 messageId。
  • 非 A2A 代理调用错误：如果代理在注册表中缺少 A2A_AGENT 协议条目（例如某些预构建或托管代理），则对其代理网址调用 getCard 会返回 501 UNIMPLEMENTED (“... 尚未受支持”)，而调用 message:send 会返回 400 INVALID_ARGUMENT (“不支持的代理”)。
  • 网址中的项目编号：注册表会返回包含项目编号而不是项目 ID 的 A2A 网址。发出 HTTP 请求时，请勿更改此数字字符串。

问题排查

请使用下表排查常见的 A2A 端点错误：

症状 可能的原因 解决方法
对 getCard 请求返回 HTTP 404 使用了不正确的路径别名（例如 /v1:getCard 或 /.well-known/agent-card.json）。 严格向 GET {url}/v1/card 发送 GET 请求。
HTTP 400 *“Unknown name 'parts'”* 使用了旧版或生成式 AI 客户端正文格式。 将文本字符串放在 content 中，而不是 parts 中。
HTTP 400 枚举值无效 role 传递了小写 "user" 或 "user_role"。 将 message.role 准确设置为 "ROLE_USER"。
HTTP 501 *“尚不支持”* 对未发布 A2A 接口的代理调用 getCard。 在调用之前，检查注册表资源的 protocols 数组以确认 A2A_AGENT 支持。
HTTP 400 *"不支持的代理"* 对非 A2A 代理调用 message:send。 选择注册表定义包含有效 A2A_AGENT 协议绑定的代理。
HTTP 401 或 HTTP 403 Permission Denied 缺少 OAuth 范围、缺少 IAM 角色或缺少配额项目标头。 检查调用方 IAM 角色（agentregistry.viewer 和 assistants.assist）；验证 cloud-platform 范围；如果使用 ADC，则传递 -H "X-Goog-User-Project: PROJECT_ID"。