解析端点和构建编排器

智能体开发套件 (ADK) 提供了一个专用 AgentRegistry 客户端，可让您以程序化方式发现、查找和连接到 Agent Registry 中编目的 AI 智能体和 Model Context Protocol (MCP) 服务器。

您可以使用 ADK 在运行时解析这些端点，而不是将端点网址硬编码到应用中。

Agent Registry 提供底层端点，但生产部署 通常通过 Agent Gateway路由这些调用。 Agent Gateway 可帮助您强制执行安全政策、执行协议调解，以及对您发现的工具应用内容过滤。

本文档介绍了如何从 Agent Registry 检索远程智能体和 MCP 工具集，并将它们包含在父级编排器智能体中。

准备工作

在将 ADK 与 Agent Registry 集成之前，请完成以下操作：

  1. 在您的项目中设置 Agent Registry。

  2. 安装 ADK 或将其升级到最新版本，并安装必要的 A2A 依赖项：

    pip 

    pip install --upgrade "google-adk[a2a]"

    uv 

    uv add "google-adk[a2a]"

    您必须至少升级到 google-adk>=1.29.0。

  3. 配置应用默认凭证 (ADC)：

    gcloud auth application-default login

ADC 凭据必须具有与智能体或工具交互的底层服务所需的 IAM 权限。您还可以选择为外部工具集使用自定义标头。如需了解详情，请参阅 对工具和资源进行身份验证。

设置环境变量

如需按照本指南操作，请设置以下环境变量：

export GOOGLE_CLOUD_PROJECT=PROJECT_ID
export GOOGLE_CLOUD_LOCATION=LOCATION

替换以下内容：

  • PROJECT_ID：您的项目 ID。
  • LOCATION：注册表区域或位置，例如 us-central1。

初始化注册表客户端

如需以编程方式与注册表交互，请使用您的项目和位置初始化 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.")

# Initialize the client
registry = AgentRegistry(
    project_id=project_id,
    location=location,
)

编写多智能体系统

ADK 抽象了底层连接机制，让您可以通过将多个专业智能体组合成灵活的层次结构来设计可伸缩的应用。

您可以使用注册表客户端提取特定资源，并将其直接传递到新 LlmAgent 智能体的定义中。您的编排器可以将远程智能体作为子智能体调用，并执行 MCP 工具，就像它们是本地 Python 函数一样。

请使用以下方法：

  • 提取远程智能体：使用 get_remote_a2a_agent()
  • 提取 MCP 工具集：使用 get_mcp_toolset()

以下示例演示了如何通过 构建编排器智能体来编写多智能体系统，该智能体利用了已注册的旅行智能体 和已注册的 Compute Engine MCP 服务器。 在此示例中，身份验证由智能体自己的身份处理，但您可以使用其他方法，例如 API 密钥和 OAuth。如需了解详情，请参阅 对工具和资源进行身份验证。

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

# Define the GoogleAuth class for the HTTP client
class GoogleAuth(httpx.Auth):
    def __init__(self):
        self.creds, _ = google.auth.default()
    def auth_flow(self, request):
        if not self.creds.valid:
            self.creds.refresh(Request())
        request.headers["Authorization"] = f"Bearer {self.creds.token}"
        yield request

# Connect to a remote A2A agent using its resource name in short or full format
# Short formats automatically imply the client's configured project and location
# Short format: "agents/AGENT_ID"
# Full format: f"projects/{project_id}/locations/{location}/agents/AGENT_ID"
agent_name = "agents/AGENT_ID"

# Configure the HTTP client with GoogleAuth and a 60-second timeout
httpx_client = httpx.AsyncClient(auth=GoogleAuth(), timeout=httpx.Timeout(60.0))
my_remote_agent = registry.get_remote_a2a_agent(
    agent_name=agent_name,
    httpx_client=httpx_client
)

# Retrieve an MCP toolset using its resource name in short or full format
# Short formats automatically imply the client's configured project and location
# Short format: "mcpServers/SERVER_ID"
# Full format: f"projects/{project_id}/locations/{location}/mcpServers/SERVER_ID"
mcp_server_name = "mcpServers/SERVER_ID"
my_mcp_toolset = registry.get_mcp_toolset(mcp_server_name=mcp_server_name)

# Compose the orchestrator agent
main_agent = LlmAgent(
    model="MODEL_ID", # Replace with a model such as gemini-1.5-flash
    name="travel_orchestrator",
    instruction="""You are a travel coordinator. You can use your
                   sub-agents to book travel and your tools to query
                   historical travel data.""",
    tools=[my_mcp_toolset],
    sub_agents=[my_remote_agent],
)

# You can now run your orchestrator agent
# response = await main_agent.run('Book a flight to Paris and check my past trips.')

重复使用智能体的最佳实践

为了最大限度地减少网络延迟时间，请在应用启动时从注册数据库提取智能体和工具集一次，而不是在每次调用时都调用 get_remote_a2a_agent()。

一个智能体一次只能有一个父智能体。如果您尝试将同一提取的智能体实例分配给多个编排器，ADK 可能会抛出错误，指出该智能体已有一个父级。

如需在多个父级智能体之间重复使用发现的智能体，请使用 .clone() 方法创建智能体对象的新实例。

以下示例展示了如何提取智能体一次，并将其克隆以供在不同的编排器中使用：

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

# Define the GoogleAuth class for the HTTP client
class GoogleAuth(httpx.Auth):
    def __init__(self):
        self.creds, _ = google.auth.default()
    def auth_flow(self, request):
        if not self.creds.valid:
            self.creds.refresh(Request())
        request.headers["Authorization"] = f"Bearer {self.creds.token}"
        yield request

# Configure the HTTP client with GoogleAuth and a 60-second timeout
httpx_client = httpx.AsyncClient(auth=GoogleAuth(), timeout=httpx.Timeout(60.0))

# Fetch the remote agent once during startup
# Use the resource name in short or full format
# Short formats automatically imply the client's configured project and location
# Short format: "agents/AGENT_ID"
# Full format: f"projects/{project_id}/locations/{location}/agents/AGENT_ID"
agent_name = f"projects/PROJECT_ID/locations/LOCATION/agents/AGENT_ID"
base_remote_agent = registry.get_remote_a2a_agent(
    agent_name=agent_name,
    httpx_client=httpx_client
)

# Use .clone() to assign the agent to different parent orchestrators
flight_orchestrator = LlmAgent(
    model="gemini-1.5-flash",
    name="flight_orchestrator",
    sub_agents=[base_remote_agent.clone()]
)

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

后续步骤