Google Cloud 智能体注册表

Supported in ADKPython v1.26.0Go v2.1.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）。
• 按照安装部分的说明安装适用于你所用语言的 ADK。

有关从 ADK 智能体连接到 Google Cloud 的更多信息，请参阅连接 Google Cloud 和 Agent Platform。

安装

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

PythonGo

pip install google-adk

必要的依赖项

google.adk.integrations.agent_registry 模块在模块作用域内同时导入了 A2A SDK 和 Agent Identity 身份验证提供者，因此仅安装核心包时导入 AgentRegistry 会抛出 ImportError。请同时安装 a2a 和 agent-identity 附加组件：

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

go get google.golang.org/adk/v2

客户端位于核心模块的 google.golang.org/adk/v2/agentregistry 包中，因此无需额外安装。

与智能体配合使用

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

PythonGo

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

# 1. 初始化
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. 列出资源
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. 使用远程 A2A 智能体
# 替换为你的已注册智能体的完整资源名称
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. 使用 MCP 工具集
# 替换为你的已注册 MCP 服务器的完整资源名称
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. 示例智能体组合
main_agent = LlmAgent(
    model="gemini-flash-latest", # 或你偏好的模型
    name="demo_agent",
    instruction="You can leverage registered tools and sub-agents.",
    tools=[my_mcp_toolset],
    sub_agents=[my_remote_agent],
)

package main

import (
    "cmp"
    "context"
    "fmt"
    "log"
    "os"

    "google.golang.org/genai"

    "google.golang.org/adk/v2/agent"
    "google.golang.org/adk/v2/agent/llmagent"
    "google.golang.org/adk/v2/agentregistry"
    "google.golang.org/adk/v2/cmd/launcher"
    "google.golang.org/adk/v2/cmd/launcher/full"
    "google.golang.org/adk/v2/model/gemini"
    "google.golang.org/adk/v2/tool"
)

func main() {
    ctx := context.Background()

    // 1. 初始化
    projectID := os.Getenv("GOOGLE_CLOUD_PROJECT")
    if projectID == "" {
        log.Fatal("GOOGLE_CLOUD_PROJECT environment variable not set.")
    }
    location := cmp.Or(os.Getenv("GOOGLE_CLOUD_LOCATION"), "global")

    registry, err := agentregistry.New(ctx, agentregistry.Config{
        ProjectID: projectID,
        Location:  location,
    })
    if err != nil {
        log.Fatalf("Failed to create the registry client: %v", err)
    }

    // 2. 列出资源。All* 迭代器按需获取分页数据，
    // 并将获取失败的页面报告为单个 (nil, error)。
    fmt.Println("Listing Agents...")
    for a, err := range registry.AllAgents(ctx) {
        if err != nil {
            log.Fatalf("Failed to list agents: %v", err)
        }
        fmt.Printf("  - %s (%s)\n", a.Name, a.DisplayName)
    }

    fmt.Println("Listing MCP Servers...")
    for s, err := range registry.AllMCPServers(ctx) {
        if err != nil {
            log.Fatalf("Failed to list MCP servers: %v", err)
        }
        fmt.Printf("  - %s (%s)\n", s.Name, s.DisplayName)
    }

    // 3. 使用远程 A2A 智能体
    // 替换为你的已注册智能体的完整资源名称
    agentName := fmt.Sprintf("projects/%s/locations/%s/agents/YOUR_AGENT_ID", projectID, location)
    myRemoteAgent, err := registry.RemoteAgent(ctx, agentName)
    if err != nil {
        log.Fatalf("Failed to resolve the remote agent: %v", err)
    }

    // 4. 使用 MCP 工具集
    // 替换为你的已注册 MCP 服务器的完整资源名称
    mcpServerName := fmt.Sprintf("projects/%s/locations/%s/mcpServers/YOUR_MCP_SERVER_ID", projectID, location)
    myMCPToolset, err := registry.MCPToolset(ctx, mcpServerName)
    if err != nil {
        log.Fatalf("Failed to connect to the MCP server: %v", err)
    }

    // 5. 示例智能体组合
    model, err := gemini.NewModel(ctx, "gemini-flash-latest", &genai.ClientConfig{})
    if err != nil {
        log.Fatalf("Failed to create the model: %v", err)
    }

    rootAgent, err := llmagent.New(llmagent.Config{
        Name:        "demo_agent",
        Model:       model,
        Instruction: "You can leverage registered tools and sub-agents.",
        Toolsets:    []tool.Toolset{myMCPToolset},
        SubAgents:   []agent.Agent{myRemoteAgent},
    })
    if err != nil {
        log.Fatalf("Failed to create the agent: %v", err)
    }

    config := &launcher.Config{AgentLoader: agent.NewSingleLoader(rootAgent)}
    l := full.NewLauncher()
    if err := l.Execute(ctx, config, os.Args[1:]); err != nil {
        log.Fatalf("Run failed: %v\n\n%s", err, l.CommandLineSyntax())
    }
}

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

远程 A2A 智能体

对远程 A2A 智能体的调用不会自动进行身份验证。如果你正在连接到 Google A2A 智能体，请在创建远程智能体时提供一个经过身份验证的 HTTP 客户端。

PythonGo

将配置了 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,
)

使用 WithA2AHTTPClient 传递经过身份验证的 *http.Client，或使用 WithA2AHeaders 传递静态头信息。

import (
    "golang.org/x/oauth2/google"

    "google.golang.org/adk/v2/agentregistry"
)

httpClient, err := google.DefaultClient(ctx, "https://www.googleapis.com/auth/cloud-platform")
if err != nil {
    log.Fatalf("Failed to load Application Default Credentials: %v", err)
}

remoteAgent, err := registry.RemoteAgent(ctx, agentName,
    agentregistry.WithA2AHTTPClient(httpClient),
)

请在客户端的 Transport 上设置超时时间，而不是使用 http.Client.Timeout，因为后者应用于整个请求，可能会截断流式响应。

Google MCP 服务器

对于 Google MCP 服务器，身份验证头会自动传递。

PythonGo

如果自动身份验证未按预期工作，你可以使用 AgentRegistry 构造函数中的 header_provider 参数手动提供头信息。

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
)

对 *.googleapis.com 端点的请求会复用注册表客户端本身的凭据。对于任何其他端点，或要覆盖该默认行为，请传入 WithMCPHTTPClient 和 WithMCPHeaders。

toolset, err := registry.MCPToolset(ctx, mcpServerName,
    agentregistry.WithMCPHTTPClient(httpClient),
    agentregistry.WithMCPHeaders(map[string]string{"X-Tenant-Id": "acme"}),
)

以这种方式设置的头信息将应用于工具集向 MCP 服务器发送的每个请求。它们不会影响对 Agent Registry API 本身的调用。

API 参考

PythonGo

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.Client 类型为每种资源类型提供三种发现方法：List* 返回单页结果，Get* 通过完整资源名称返回单个资源，All* 返回一个按需获取分页数据的 iter.Seq2。

• ListAgents(ctx, opts ...ListOption)、GetAgent(ctx, name)、AllAgents(ctx, opts ...ListOption)：已注册的 A2A 智能体。
• ListMCPServers(ctx, opts ...ListOption)、GetMCPServer(ctx, name)、AllMCPServers(ctx, opts ...ListOption)：已注册的 MCP 服务器。
• ListEndpoints(ctx, opts ...ListOption)、GetEndpoint(ctx, name)、AllEndpoints(ctx, opts ...ListOption)：已注册的模型端点。
• RemoteAgent(ctx, name, opts ...RemoteAgentOption)：将已注册的 A2A 智能体解析为可用作子智能体的 agent.Agent。
• MCPToolset(ctx, name, opts ...MCPToolsetOption)：将已注册的 MCP 服务器解析为 tool.Toolset。

列表选项包括 WithFilter、WithPageSize 和 WithPageToken。All* 迭代器会自行管理分页令牌。来自 Agent Registry API 的非 2xx 响应将以 *agentregistry.APIError 的形式返回，其中包含 StatusCode 和响应 Body。

配置选项

PythonGo

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

• project_id（str，必填）：Google Cloud 项目 ID。
• location（str，必填）：Google Cloud 位置/区域，例如 "global"、"us-central1"。
• header_provider（Callable，可选）：一个可调用对象，接受 ReadonlyContext 并返回一个自定义头信息字典。这些头信息将包含在 get_mcp_toolset 返回的 McpToolset 对目标 MCP 服务器发出的请求中。这些头信息不会影响对 Agent Registry API 本身的调用，也不会影响 RemoteA2aAgent 发出的请求。对于这些请求，请将经过身份验证的 httpx.AsyncClient 传递给 get_remote_a2a_agent，如远程 A2A 智能体部分所示。

agentregistry.New 构造函数接受一个 Config 结构体：

• ProjectID（string，必填）：Google Cloud 项目 ID。
• Location（string，必填）：Google Cloud 位置/区域，例如 "global"、"us-central1"。
• HTTPClient（*http.Client，可选）：用于 Agent Registry API 调用的客户端。当为 nil 时，ADK 会从应用程序默认凭据构建一个客户端，并从 GOOGLE_API_USE_MTLS_ENDPOINT 和 GOOGLE_API_USE_CLIENT_CERTIFICATE 解析端点（包括 mTLS）。此客户端也会复用于到 *.googleapis.com 端点的 McpToolset 流量，但不会用于 A2A 流量。

到已解析端点的出站连接则按调用分别配置：RemoteAgent 使用 WithA2AHTTPClient 和 WithA2AHeaders，MCPToolset 使用 WithMCPHTTPClient 和 WithMCPHeaders。

附加资源

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