解析端点和构建编排器

智能体开发套件 (ADK) 提供了一个专用的 AgentRegistry 客户端，可让您以编程方式发现、查找和连接到在 Agent Registry 中注册的 AI 智能体和 Model Context Protocol (MCP) 服务器。您可以使用 ADK 在运行时解析这些端点，并将其组合到父级编排器智能体中，而不是将端点网址硬编码到应用中。

Agent Registry 提供底层端点。对于生产部署，您可以通过 Agent Gateway 路由调用，以强制执行安全政策、执行协议中介和应用内容过滤。

准备工作

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

  1. 设置 Agent Registry。

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

  3. 安装或升级到最新版本的 ADK，其中包含 A2A、智能体身份和 MCP 依赖项：

    pip 

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

    uv 

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

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

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

    gcloud auth application-default login

您的凭据必须具有代理或工具与之交互的基础服务所需的 Identity and Access Management (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 智能体的定义中。您的编排器可以调用远程智能体作为子智能体，并像执行本地 Python 函数一样执行 MCP 工具。

如需提取远程智能体或 MCP 工具集，请在 Python 智能体代码中调用以下 AgentRegistry 客户端方法：

  • 如需提取远程代理：请使用 get_remote_a2a_agent()。
  • 提取 MCP 工具集：使用 get_mcp_toolset()。

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

如需使用可视化低代码界面设计和编排代理，请直接在 Gemini Enterprise Agent Platform 画布上的 Agent Studio 中配置已注册的子代理。如需了解详情，请参阅在 Agent Studio 中设计代理。

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
                   subagents 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.')

替换以下内容：

  • AGENT_ID：已注册代理的 ID。
  • SERVER_ID：已注册的 MCP 服务器的 ID。
  • MODEL_ID：编排器智能体的模型 ID，例如 gemini-1.5-flash。

重复使用代理的最佳实践

为最大限度地减少网络延迟，请在应用启动时从注册表中提取一次智能体和工具集，而不是在每次调用时都调用 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()]
)

替换以下内容：

  • PROJECT_ID：治理项目的项目 ID。
  • LOCATION：注册表区域或位置，例如 us-central1。
  • AGENT_ID：已注册代理的 ID。

后续步骤