Google 会使用 AI 技术将内容翻译成您偏好的语言。AI 翻译可能包含错误。

构建多智能体系统并调用 A2A 智能体

在多智能体系统中，编排智能体会将任务分派给专门的目标智能体和工具。这些目标智能体将成为多智能体系统或应用的子智能体。

借助 Agent Registry，您的编排器智能体可以在运行时动态查找 Agent2Agent (A2A) 和 Model Context Protocol (MCP) 端点，而无需在应用代码中对服务网址进行硬编码。您可以使用智能体开发套件 (ADK) 中的 AgentRegistry 客户端或标准 HTTP 和 google-auth 库，检索已注册的智能体和 MCP 服务器，以构建编排器智能体。

当编排器代理调用目标代理的端点时，它会将请求直接发送到目标代理运行时或 MCP 服务器。如需在目标智能体端点上强制执行安全政策、执行协议中介和应用内容过滤，您可以通过 Agent Gateway 路由智能体调用。

准备工作

在通过 Agent Registry 查找目标智能体端点网址或调用 A2A 智能体之前，请完成以下步骤：

设置 Agent Registry。
对于单项目设置，请获取已启用 Agent Registry 的治理项目的项目 ID。如果您为多个项目配置了管理边界，请使用管理项目的 ID 进行边界范围内的发现和注册，或指定成员项目 ID 进行单项目操作。

为您的框架安装客户端依赖项：

ADK (pip)

pip install --upgrade "google-adk[a2a,agent-identity,mcp]"

您必须至少安装 google-adk>=1.29.0。

ADK（uv）

uv add "google-adk[a2a,agent-identity,mcp]"

您必须至少安装 google-adk>=1.29.0。

非 ADK Python

pip install --upgrade google-auth requests httpx

为本地开发配置应用默认凭证 (ADC)。否则，请在外部环境中附加服务账号或工作负载身份联合凭据配置：
```
gcloud auth application-default login
```
设置以下环境变量：
```
export GOOGLE_CLOUD_PROJECT=PROJECT_ID
export GOOGLE_CLOUD_LOCATION=LOCATION
```
替换以下内容：
- PROJECT_ID：您的治理项目的项目 ID。
- LOCATION：注册智能体和 MCP 服务器的智能体注册表位置，例如 global 或 us-central1。

所需的角色

您可以使用用户账号进行本地开发，也可以使用服务账号或智能体主体，让编排器智能体发现已注册的组件并调用目标 A2A 智能体。用户账号、服务账号或智能体主体必须具有适当的角色，才能发现组件并调用目标 A2A 智能体。

为确保您的用户账号、服务账号或智能体主体拥有发现已注册组件和调用目标 A2A 智能体所需的权限，请让管理员在启用了 Agent Registry 的治理项目中，向您的用户账号、服务账号或智能体主体授予以下 IAM 角色：

如需发现已注册的智能体和 MCP 服务器，请使用：Agent Registry API Viewer (roles/agentregistry.viewer)
在运行时调用目标 A2A 智能体：托管目标智能体的服务的运行时或网关角色，调用该智能体时需要此角色。例如，如果目标智能体在 Cloud Run 上运行，则账号或主账号必须具有 Cloud Run Invoker 角色 (roles/run.invoker)。

如需详细了解如何授予角色，请参阅管理对项目、文件夹和组织的访问权限。

您的管理员还可以通过自定义角色或其他预定义角色向您的用户账号、服务账号或智能体主体授予所需的权限。

为目标 A2A 智能体端点选择凭证

在 Agent Registry 中查找组件和调用其目标 Agent Endpoint 需要单独的身份验证令牌。当编排器智能体调用目标 A2A 智能体端点时，必须附加该端点预期的不记名令牌类型：

目标 A2A 智能体端点	必需的不记名令牌	凭证来源和指南
Google Cloud API 代理端点 (`https://*.googleapis.com`)	具有 `https://www.googleapis.com/auth/cloud-platform` 范围或更窄服务范围的 OAuth 2.0 访问令牌。使用未配置默认配额项目的应用默认凭据 (ADC) 时，请添加 `X-Goog-User-Project` 标头，以指定要为 API 配额付费的项目。如需了解详情，请参阅通过 REST 请求设置配额项目。	`google.auth.default()`，使用应用默认凭证 (ADC)、关联的服务账号或工作负载身份联合凭证配置。如需限制每次调用的权限，请参阅使用服务账号。
Cloud Run 或 Identity-Aware Proxy (IAP) HTTPS 代理端点（`https://*.run.app` 或自定义网域）	ID 令牌，其 `aud` 声明与目标服务网址或 IAP OAuth 2.0 客户端 ID 匹配。	`google.oauth2.id_token.fetch_id_token()`。如需了解设置说明，请参阅获取 ID 令牌、Cloud Run 上的服务到服务身份验证和IAP 的程序化身份验证。
绑定了 OAuth 的外部代理或 SaaS 代理	2-legged OAuth 2.0 令牌、3-legged OAuth 2.0 令牌或 API 密钥。	当 Agent Identity 身份验证管理器绑定附加到代理时，在 ADK 中自动管理。如需了解设置说明，请参阅向工具和资源进行身份验证。

在 ADK 中初始化 `AgentRegistry` 客户端

如需通过 ADK 与 Agent Registry 交互，请在 Python 代码中使用治理项目 ID 和 Agent Registry 位置初始化 AgentRegistry 客户端：

import os
from google.adk.integrations.agent_registry import AgentRegistry

project_id = os.environ.get("GOOGLE_CLOUD_PROJECT")
location = os.environ.get("GOOGLE_CLOUD_LOCATION", "global")

if not project_id:
    raise ValueError("GOOGLE_CLOUD_PROJECT environment variable not set.")

registry = AgentRegistry(
    project_id=project_id,
    location=location,
)

初始化 ADK AgentRegistry 客户端后，您可以从 Agent Registry 检索远程 A2A 智能体和远程 MCP 工具。在 ADK 中，已注册的 A2A 智能体会被封装为 RemoteA2aAgent 实例。这些实例称为远程代理，因为 ADK 会通过网络将任务委托给它们，而不是在本地进程中运行它们。同样，已注册的远程 MCP 服务器会封装为 McpToolset 实例。

使用 ADK 编写多智能体系统

如需从 Agent Registry 检索远程 A2A 子智能体和远程 MCP 工具集，并将其包含在 LlmAgent 编排器智能体定义中，请使用初始化的 AgentRegistry 客户端：

如需检索远程 A2A 智能体：请使用子智能体的相对资源名称调用 registry.get_remote_a2a_agent()。此资源名称可以是格式为 agents/AGENT_ID 的相对资源名称（使用在 AgentRegistry 客户端上配置的项目 ID 和位置），也可以是格式为 projects/PROJECT_ID/locations/LOCATION/agents/AGENT_ID 的完整资源名称。如果代理存在身份验证管理器 Binding，get_remote_a2a_agent() 会自动配置请求凭据。如果不存在 Binding，请传递使用 httpx.Auth 处理程序配置的自定义 httpx_client。
如需检索远程 MCP 工具集：请使用 MCP 服务器的相对资源名称 (mcpServers/SERVER_ID) 或完整资源名称 (projects/PROJECT_ID/locations/LOCATION/mcpServers/SERVER_ID) 调用 registry.get_mcp_toolset()。

以下示例展示了如何在 ADK agent.py 文件中定义编排器智能体。在此示例中，请考虑以下事项：

GoogleApiAuth 类实现了一个 httpx.Auth 处理程序，用于调用Google Cloud API (https://*.googleapis.com) A2A 子代理。它使用 google.auth.default() 加载您的应用默认凭据、附加的服务账号或工作负载身份联合凭证，并在每个请求中设置 Authorization 和 X-Goog-User-Project 标头。

注意：https://www.googleapis.com/auth/cloud-platform 范围是 Google Cloud API 所需的标准 OAuth 2.0 范围，而精细访问权限由 IAM 角色控制。如需限制范围或为每个子代理模拟专用服务账号，请参阅服务账号模拟。如果目标子智能体在 Cloud Run 上运行或位于 IAP 后面，请改为附加 OIDC ID 令牌，如为目标 A2A 智能体端点选择凭证中所述。
registry 客户端从 Agent Registry 检索远程 A2A 子智能体和远程 MCP 工具集，并通过 sub_agents 和 tools 参数将其传递给 LlmAgent。

import httpx
import google.auth
from google.auth.transport.requests import Request
from google.adk.agents.llm_agent import LlmAgent

class GoogleApiAuth(httpx.Auth):
    def __init__(self, user_project_id: str):
        self.user_project_id = user_project_id
        self.creds, _ = google.auth.default(
            scopes=["https://www.googleapis.com/auth/cloud-platform"]
        )

    def auth_flow(self, request):
        if not self.creds.valid:
            self.creds.refresh(Request())
        request.headers["Authorization"] = f"Bearer {self.creds.token}"
        request.headers["X-Goog-User-Project"] = self.user_project_id
        yield request

quota_project_id = "QUOTA_PROJECT_ID"
agent_name = "agents/AGENT_ID"
httpx_client = httpx.AsyncClient(
    auth=GoogleApiAuth(quota_project_id),
    timeout=httpx.Timeout(60.0),
)
my_remote_agent = registry.get_remote_a2a_agent(
    agent_name=agent_name,
    httpx_client=httpx_client,
)

mcp_server_name = "mcpServers/SERVER_ID"
my_mcp_toolset = registry.get_mcp_toolset(mcp_server_name=mcp_server_name)

root_agent = LlmAgent(
    model="MODEL_ID",
    name="travel_orchestrator",
    instruction=(
        "You are a travel coordinator. Use your subagents to book travel "
        "and your tools to query historical travel data."
    ),
    tools=[my_mcp_toolset],
    sub_agents=[my_remote_agent],
)

替换以下内容：

QUOTA_PROJECT_ID：用于对Google Cloud API 代理端点的请求进行配额和结算归因的项目 ID。
AGENT_ID：您希望编排器代理调用的已注册 A2A 子代理的 ID，例如 agentregistry-01234456-7890-abcd-1234-567890abcdef。
SERVER_ID：您希望编排器代理使用的已注册 MCP 服务器的 ID。
MODEL_ID：编排器代理的模型 ID，例如 gemini-2.5-flash。

在 ADK 项目的 agent.py 文件中保存 root_agent 后，您可以使用 adk run 或 adk web CLI 命令在本地测试和运行编排器智能体，也可以将其部署到 Gemini Enterprise Agent Platform 上的 Agent Runtime。如需了解完整的项目结构和执行说明，请参阅 ADK Python 快速入门和 GitHub 上的 ADK 智能体注册表示例。

从非 ADK 客户端调用 A2A 智能体

如果您在不使用 ADK 的情况下构建自定义编排器代理或客户端，则应用会执行两个经过身份验证的步骤：

发现：使用 OAuth 2.0 访问令牌向 Agent Registry API 进行身份验证，以检索目标子代理的 Agent 资源并检查其代理卡片。
运行时调用：附加目标 Agent Endpoint 所需的不记名令牌，该令牌是 *.googleapis.com Agent Endpoint 的 OAuth 2.0 访问令牌，或 Cloud Run 和 IAP Agent Endpoint 的 OIDC ID 令牌。然后，发送 A2A SendMessage 请求。输出会返回一个 SendMessageResponse JSON 对象，其中包含格式为 {"task": ...} 或 {"message": ...} 的 task 或 message 字段。此字段包含子智能体的回答，编排器智能体可以解析该回答并在其工作流中使用。

如需使用 google-auth 和 httpx 查找并调用已注册的 A2A 子代理，请完成以下步骤：

使用 OAuth 2.0 访问令牌向 Agent Registry 进行身份验证，并从 Agent Registry API 提取子代理的 Agent 资源：

import os
import uuid
from urllib.parse import urlparse
import google.auth
from google.auth.transport.requests import Request
from google.oauth2 import id_token
import httpx

project_id = os.environ["GOOGLE_CLOUD_PROJECT"]
location = os.environ.get("GOOGLE_CLOUD_LOCATION", "global")
agent_id = "AGENT_ID"

creds, _ = google.auth.default(
    scopes=["https://www.googleapis.com/auth/cloud-platform"]
)
creds.refresh(Request())

registry_url = (
    f"https://agentregistry.googleapis.com/v1/projects/{project_id}"
    f"/locations/{location}/agents/{agent_id}"
)
client = httpx.Client(timeout=60.0)
reg_response = client.get(
    registry_url,
    headers={
        "Authorization": f"Bearer {creds.token}",
        "X-Goog-User-Project": project_id,
    },
)
reg_response.raise_for_status()
agent_resource = reg_response.json()

将 AGENT_ID 替换为您要调用的已注册 A2A 子代理的 ID，例如 agentregistry-01234456-7890-abcd-1234-567890abcdef。

从 Agent 资源中的智能体卡片中提取基于 HTTP 的 A2A 接口：

card = agent_resource.get("card", {}).get("content", {})
interfaces = card.get("supportedInterfaces") or (
    (agent_resource.get("protocols") or [{}])[0].get("interfaces", [])
)
iface = next(
    i for i in interfaces
    if i.get("protocolBinding", "HTTP+JSON")
    in ("HTTP+JSON", "HTTP_JSON", "JSONRPC")
)
endpoint_url = iface["url"]
binding = iface.get("protocolBinding", "HTTP+JSON")
tenant = iface.get("tenant", "")

使用 OAuth 2.0 访问令牌（适用于 *.googleapis.com Agent Endpoint）或 OIDC ID 令牌（适用于 Cloud Run 或 IAP Agent Endpoint）准备请求标头：

host = urlparse(endpoint_url).hostname or ""
headers = {"Content-Type": "application/json", "A2A-Version": "1.0"}
if host == "googleapis.com" or host.endswith(".googleapis.com"):
    headers["Authorization"] = f"Bearer {creds.token}"
    headers["X-Goog-User-Project"] = project_id
else:
    oidc_audience = f"https://{host}"
    oidc_jwt = id_token.fetch_id_token(Request(), oidc_audience)
    headers["Authorization"] = f"Bearer {oidc_jwt}"

使用智能体卡片中的协议绑定，设置 A2A SendMessage 请求的格式并发送该请求，然后解析子智能体的响应：

message = {
    "messageId": str(uuid.uuid4()),
    "role": "ROLE_USER",
    "parts": [{"text": "Summarize recent deployment status."}],
}
if binding == "JSONRPC":
    params = {"message": message}
    if tenant:
        params["tenant"] = tenant
    payload = {
        "jsonrpc": "2.0",
        "id": str(uuid.uuid4()),
        "method": "SendMessage",
        "params": params,
    }
    target_url = endpoint_url
else:
    payload = {"message": message}
    prefix = f"/{tenant.strip('/')}" if tenant else ""
    target_url = f"{endpoint_url.rstrip('/')}{prefix}/message:send"

response = client.post(target_url, headers=headers, json=payload)
response.raise_for_status()
body = response.json()
if binding == "JSONRPC":
    if "error" in body:
        raise RuntimeError(f"JSON-RPC error: {body['error']}")
    send_message_response = body["result"]
else:
    send_message_response = body
print(send_message_response)
client.close()

此步骤会为 A2A 版本 1.0 设置 SendMessage 请求的格式。如果您需要在 A2A protocol 版本 1.0 和 0.3 之间自动协商，请使用 ADK 或 a2a-sdk Python 客户端。

在 ADK 中跨编排器智能体重复使用远程智能体

为最大限度地减少网络延迟，请在应用启动时从 Agent Registry 中检索一次远程智能体和工具集，而不是在每次请求时都调用 get_remote_a2a_agent()。

在 ADK 中，一个 RemoteA2aAgent 实例一次只能有一个父级编排器智能体。继续初始化设置，调用 .clone() 为每个父级编排器智能体创建一个独立副本：

agent_path = (
    f"projects/{project_id}/locations/{location}"
    "/agents/AGENT_ID"
)
base_remote_agent = registry.get_remote_a2a_agent(
    agent_name=agent_path,
    httpx_client=httpx_client,
)

flight_orchestrator = LlmAgent(
    model="gemini-2.5-flash",
    name="flight_orchestrator",
    sub_agents=[base_remote_agent.clone()],
)

hotel_orchestrator = LlmAgent(
    model="gemini-2.5-flash",
    name="hotel_orchestrator",
    sub_agents=[base_remote_agent.clone()],
)

将 AGENT_ID 替换为您希望编排器代理调用的已注册 A2A 子代理的 ID。

后续步骤

查看 A2A protocol，了解传输绑定和任务生命周期状态。
了解如何设置 Agent Gateway，以便为目标代理端点路由和治理流量。
了解如何向工具和资源进行身份验证，以及如何在代理调用之间限定凭据范围。
了解如何在 Agent Registry 中注册 skill 和管理 skill。