Google Cloud Agent Registry

Supported in ADKPython v1.26.0Preview

Agent Development Kit (ADK) 中的 Agent Registry 客户端库允许开发者发现、查找和连接到 Google Cloud Agent Registry 中编目的 AI 智能体和 MCP 服务器。这使得能够使用受治理的组件动态组合基于智能体的应用程序。

使用场景

• 加速开发：从中央目录轻松查找和重用现有智能体和工具（MCP 服务器），而无需重新构建它们。
• 动态集成：在运行时发现智能体和 MCP 服务器端点，使应用程序对环境变化更加稳健。
• 增强治理：在 ADK 应用程序中使用来自注册表的受治理和已验证的组件。

前置条件

• 一个 Google Cloud 项目。
• 在你的 Google Cloud 项目中启用了 Agent Registry API。
• 为你的环境配置了身份验证。你应该使用应用程序默认凭据 登录（gcloud auth application-default login）。
• 环境变量 GOOGLE_CLOUD_PROJECT 设置为你的项目 ID，GOOGLE_CLOUD_LOCATION 设置为适当的区域（例如 global、us-central1）。
• 已安装 google-adk 库。

安装

Agent Registry 集成是核心 ADK 库的一部分。

pip install google-adk

可选依赖

要使用 AgentRegistry 集成的全部功能，你可能需要根据你的用例安装额外的附加组件：

对于 A2A（智能体间）支持： 如果你计划使用 get_remote_a2a_agent 或与远程 A2A 兼容的智能体交互，请安装 a2a 附加组件：

pip install "google-adk[a2a]"

对于智能体身份（GCP 认证提供者）： 如果你需要使用 GcpAuthProvider（例如，当 get_mcp_toolset 通过 IAM 绑定自动解析已注册 MCP 服务器的身份验证时），请安装 agent-identity 附加组件：

pip install "google-adk[agent-identity]"

与智能体配合使用

在 ADK 智能体中使用 Agent Registry 集成的主要方式是使用 AgentRegistry 客户端动态获取远程智能体或工具集。

from google.adk.agents.llm_agent import LlmAgent
from google.adk.integrations.agent_registry import AgentRegistry
import os

# 1. Initialization
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,
)

# 2. Listing Resources
print("Listing Agents...")
agents_response = registry.list_agents()
for agent in agents_response.get("agents", []):
    print(f"  - {agent.get('name')} ({agent.get('displayName')})")

print("Listing MCP Servers...")
mcp_servers_response = registry.list_mcp_servers()
for server in mcp_servers_response.get("mcpServers", []):
    print(f"  - {server.get('name')} ({server.get('displayName')})")

# 3. Using a Remote A2A Agent
# Replace with the full resource name of your registered agent
agent_name = f"projects/{project_id}/locations/{location}/agents/YOUR_AGENT_ID"
my_remote_agent = registry.get_remote_a2a_agent(agent_name=agent_name)

# 4. Using an MCP Toolset
# Replace with the full resource name of your registered MCP server
mcp_server_name = f"projects/{project_id}/locations/{location}/mcpServers/YOUR_MCP_SERVER_ID"
my_mcp_toolset = registry.get_mcp_toolset(mcp_server_name=mcp_server_name)

# 5. Example Agent Composition
main_agent = LlmAgent(
    model="gemini-flash-latest", # Or your preferred model
    name="demo_agent",
    instruction="You can leverage registered tools and sub-agents.",
    tools=[my_mcp_toolset],
    sub_agents=[my_remote_agent],
)

Google MCP 服务器和远程 A2A 智能体的身份验证

远程 A2A 智能体

如果你要连接到 Google A2A 智能体，需要将配置了 Google 身份验证标头的 httpx.AsyncClient 传递给 get_remote_a2a_agent 方法。

示例：

import httpx
import google.auth
from google.auth.transport.requests import Request

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

httpx_client = httpx.AsyncClient(auth=GoogleAuth(), timeout=httpx.Timeout(60.0))
remote_agent = registry.get_remote_a2a_agent(
    f"projects/{project_id}/locations/{location}/agents/YOUR_AGENT_ID",
    httpx_client=httpx_client,
)

Google MCP 服务器

对于 Google MCP 服务器，身份验证标头会自动传递。但是，如果自动身份验证未按预期工作，你可以使用 AgentRegistry 构造函数中的 header_provider 参数手动提供标头。

Example:

import google.auth
from google.auth.transport.requests import Request
from google.adk.integrations.agent_registry import AgentRegistry

def google_auth_header_provider(context):
    creds, _ = google.auth.default()
    if not creds.valid:
        creds.refresh(Request())
    return {"Authorization": f"Bearer {creds.token}"}

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

API 参考

AgentRegistry 类提供以下核心方法：

• list_mcp_servers(self, filter_str, page_size, page_token): 获取已注册 MCP 服务器的列表。
• get_mcp_server(self, name): 检索特定 MCP 服务器的详细元数据。
• get_mcp_toolset(self, mcp_server_name): 从已注册的 MCP 服务器构建 ADK McpToolset 实例。
• list_agents(self, filter_str, page_size, page_token): 获取已注册 A2A 智能体的列表。
• get_agent_info(self, name): 检索特定 A2A 智能体的详细元数据。
• get_remote_a2a_agent(self, agent_name): 为已注册的 A2A 智能体创建 ADK RemoteA2aAgent 实例。

配置选项

AgentRegistry 构造函数接受以下参数：

• project_id (str, 必填): Google Cloud 项目 ID。
• location (str, 必填): Google Cloud 位置/区域，例如 "global"、"us-central1"。
• header_provider (Callable, 可选): 一个可调用对象，接收 ReadonlyContext 并返回自定义标头字典，这些标头将包含在 McpToolset 或 RemoteA2aAgent 对目标服务发出的请求中。这不影响用于调用 Agent Registry API 本身的标头。

其他资源

• 示例智能体代码
• Agent Registry 客户端
• Google Auth 库