使用 ADK 的 AgentRegistry 集成 Google Cloud 服务：以 BigQuery MCP 工具为例的完整实战指南-平芜编程栈

使用 ADK 的 AgentRegistry 集成 Google Cloud 服务：以 BigQuery MCP 工具为例的完整实战指南

【免费下载链接】adk-pythonAn open-source, code-first Python toolkit for building, evaluating, and deploying sophisticated AI agents with flexibility and control.项目地址: https://gitcode.com/GitHub_Trending/ad/adk-python

导读

本篇文章围绕 ADK（Agent Development Kit）开源仓库中的api_registry_agent示例展开，讲解如何通过AgentRegistry客户端发现 Google Cloud Agent Registry 中注册的 MCP Server，并将其暴露的 BigQuery 工具直接挂载到LlmAgent上，让 Agent 具备真实查询 BigQuery 数据的能力。读完本文，你将掌握AgentRegistry的安装前提、配置要点、CLI / Web UI 两种运行方式，以及其底层连接解析、认证与遥测的实现原理。

示例概览：一个能真实查询 BigQuery 的数据分析 Agent

示例位于 contributing/samples/integrations/api_registry_agent/，其核心目的在 README.md 中明确说明：演示如何使用AgentRegistry发现并交互 Google Cloud 服务（如 BigQuery）——具体方式是经由一个在 Agent Registry 中注册的、暴露 BigQuery 工具的 MCP Server。

整个示例只有三个文件，职责非常清晰：

文件	作用
agent.py	核心代码：实例化`AgentRegistry`、获取 MCP Toolset、组装`LlmAgent`
README.md	运行说明：前置条件、配置项、CLI 与 Web UI 启动方式
init.py	包标记文件

从架构上看，它体现了一条典型的"注册中心发现 + 工具动态挂载"链路：Agent Registry 充当服务目录，ADK 通过AgentRegistry客户端读取注册信息，自动解析出 MCP 服务器的连接地址与认证方式，再把工具以标准 Toolset 的形式注入 Agent。

前置条件：安装、项目与 MCP Server 注册

按照 README.md 的 Prerequisites 章节，运行该示例需要满足三个条件：

1. ADK 必须安装 A2A 扩展

pip install "google-adk[a2a]"

这一点不是可选项而是硬性要求，README 给出了明确原因：AgentRegistry在模块加载时（agent.py中的 import 阶段）就会导入a2a-sdk，因此只执行普通的pip install google-adk会在导入处直接报错。对应实现可以在 agent_registry.py 中看到：

try: from a2a.types import AgentSkill from google.adk.a2a import _compat from google.adk.agents.remote_a2a_agent import RemoteA2aAgent except ImportError as e: raise ImportError( "AgentRegistry requires the 'a2a-sdk' package. " "Please install it using 'pip install google-adk[a2a]'." ) from e

在 pyproject.toml 中，a2a可选依赖被定义为a2a-sdk[http-server]>=0.3.4,<2，这也是实际安装时会拉取的版本范围。

2. 一个已启用 Agent Registry API 的 Google Cloud 项目

AgentRegistry客户端在初始化时会调用google.auth.default()获取默认凭证（见 agent_registry.py），因此运行环境需要具备可用的 Google Cloud 默认凭证（如GOOGLE_APPLICATION_CREDENTIALS环境变量或已登录的 ADC）。

3. 一个已在 Agent Registry 中注册、暴露 BigQuery 工具的 MCP Server

这是本示例能够工作的数据来源。Agent Registry 负责保存该 MCP Server 的连接元数据（endpoint URL、协议绑定、认证绑定等），而AgentRegistry客户端负责在运行时读取并解析这些元数据。

配置与代码解读

修改三个核心配置项

打开 agent.py，需要替换的是顶部的三个常量：

# TODO: Fill in with your GCloud project id and MCP server name PROJECT_ID = "your-google-cloud-project-id" LOCATION = "global" MCP_SERVER_NAME = "your-mcp-server-name"

常量	含义	示例值
`PROJECT_ID`	你的 Google Cloud 项目 ID	`my-gcp-project`
`LOCATION`	资源所在区域，示例默认使用`global`	`global`、`us-central1`等
`MCP_SERVER_NAME`	已在 Agent Registry 注册的 MCP Server 名称	`bigquery-mcp-server`

核心调用链：注册、发现、挂载

替换完配置后，代码的核心逻辑只有三行：

registry = AgentRegistry(project_id=PROJECT_ID, location=LOCATION) registry_tools = registry.get_mcp_toolset( f"projects/{PROJECT_ID}/locations/{LOCATION}/mcpServers/{MCP_SERVER_NAME}" ) root_agent = LlmAgent( name="bigquery_assistant", instruction="""...""", tools=[registry_tools], )

这里的关键路径是：

创建AgentRegistry客户端：传入project_id与location，二者缺一不可。从 agent_registry.py 的源码看，若任一参数缺失会抛出ValueError("project_id and location must be provided")，对应的测试用例是 test_agent_registry.py 中的test_init_raises_value_error_if_params_missing。
通过资源全名获取 MCP Toolset：get_mcp_toolset接收的正是 MCP Server 的完整资源名（resource name），格式为projects/{project_id}/locations/{location}/mcpServers/{server_name}。
把 Toolset 注入LlmAgent：tools=[registry_tools]将该 MCP Server 暴露的全部工具（BigQuery 查询、表结构查看等）作为 Agent 的工具集。

Agent 指令设计：引导模型"先探查、再查询"

示例中的 Agent 名叫bigquery_assistant，其instruction是一段精心设计的数据分析师提示词，值得借鉴：

You are a helpful data analyst assistant with access to BigQuery. The project ID is: {PROJECT_ID} When users ask about data: - Use the project ID {PROJECT_ID} when calling BigQuery tools. - First, explore available datasets and tables to understand what data exists. - Check table schemas to understand the structure before querying. - Write clear, efficient SQL queries to answer their questions. - Explain your findings in simple, non-technical language. Mandatory Requirements: - Always use the BigQuery tools to fetch real data rather than making assumptions. - For all BigQuery operations, use project_id: {PROJECT_ID}.

这段提示词通过动态字符串拼接注入了真实项目 ID，并明确要求 Agent：先探索数据集与表、再检查表结构、最后编写 SQL 查询，且必须调用真实工具而非凭空假设。这能有效减少 LLM 幻觉，是"工具型数据分析 Agent"的通用设计范式。

两种运行方式

方式一：CLI 运行

adk run --log_level DEBUG contributing/samples/integrations/api_registry_agent

--log_level DEBUG会输出详细日志，便于排查 MCP 连接、工具发现等环节的问题。这里传入的路径是示例目录本身（包含agent.py），ADK 会自动发现并加载该 Agent。

方式二：Web UI 运行

adk web contributing/samples/integrations

然后打开浏览器访问http://127.0.0.1:8000，在 Agent 列表中选择api_registry_agent即可开始对话。Web UI 方式启动的是整个integrations目录，因此可以看到该目录下的多个示例 Agent。

源码纵深：`AgentRegistry`客户端是如何工作的

AgentRegistry是google.adk.integrations.agent_registry模块提供的高层客户端，实现在 agent_registry.py。与标准 REST 客户端库不同，它面向 ADK 集成提供了两个关键辅助方法：get_mcp_toolset（将注册的 MCP Server 转为 Toolset）和get_remote_a2a_agent（将注册的 A2A Agent 转为可调用的远程 Agent），二者都会自动解析连接细节并处理认证。

初始化阶段：凭证、会话与 mTLS

构造函数会执行以下动作（agent_registry.py）：

调用google.auth.default()获取默认凭证，失败时抛出RuntimeError；
用requests_auth.AuthorizedSession包装凭证，全程复用同一个会话；
根据客户端证书与GOOGLE_API_USE_MTLS_ENDPOINT环境变量决定是否使用 mTLS 端点（普通端点为https://agentregistry.googleapis.com/v1，mTLS 端点为https://agentregistry.mtls.googleapis.com/v1）。

`get_mcp_toolset`：三步解析出可直接使用的 Toolset

get_mcp_toolset（agent_registry.py）的解析流程可分为三步：

读取服务器详情并解析 endpoint URL：先调用get_mcp_server获取注册信息；随后在_get_connection_uri中按协议绑定筛选，优先选择JSONRPC绑定，找不到再回退到HTTP_JSON；若两者都取不到 endpoint URI，则抛出ValueError。测试 test_agent_registry.py 验证了接口元数据（interfaces）的解析逻辑。
自动解析认证方案：如果调用方没有显式传入auth_scheme，客户端会请求bindings接口，查找该 MCP Server 的 IAM 认证绑定，并构造GcpAuthProviderScheme（见_resolve_auth_provider_scheme，agent_registry.py）。绑定查询失败时只记录警告、不会中断流程。
构建带认证与遥测信息的 Toolset：最终返回的是AgentRegistrySingleMcpToolset——它是McpToolset的子类，在构造时接收destination_resource_id，并在get_tools()中为每个工具注入GCP_MCP_SERVER_DESTINATION_ID自定义元数据（agent_registry.py），该标识会写入execute_toolspan，用于遥测追踪。对应的测试test_get_mcp_toolset_adds_destination_id（test_agent_registry.py）断言了每个工具都携带了该 MCP Server ID。

请求头注入：Google API 端点自动附加 Bearer Token

get_mcp_toolset内部还构建了一个组合式header_provider（agent_registry.py）：当没有显式认证方案且endpoint 是*.googleapis.com的 HTTPS 地址时，自动通过_get_auth_headers()刷新凭证并注入Authorization: Bearer <token>；如果调用方提供了自定义header_provider，则在此基础上叠加。这一行为在test_get_mcp_toolset_auth_headers（test_agent_registry.py）中有参数化验证。此外，所有 API 请求都会携带x-goog-api-client/user-agent中的google-adk/标识（test_registry_requests_identify_adk）。

其他可用方法

除了本示例用到的get_mcp_toolset，AgentRegistry还提供了一组可用于服务发现的方法：

MCP Server 相关：list_mcp_servers、search_mcp_servers、get_mcp_server；
Endpoint 相关：list_endpoints、get_endpoint、get_model_name；
A2A Agent 相关：list_agents、search_agents、get_agent_info、get_remote_a2a_agent。

其中get_remote_a2a_agent与get_mcp_toolset遵循相同的设计：自动从注册信息解析连接 URI 与认证绑定，返回一个开箱即用的RemoteA2aAgent（支持直接使用注册时发布的完整 Agent Card，见 agent_registry.py）。

测试与注意事项

该示例的离线加载限制

在 tests/unittests/test_samples.py 的SKIP_LOAD集合中，integrations/api_registry_agent被标记为"calls Cloud API Registry at import"，即该示例在 import 阶段就会发起对云端注册服务的调用，因此无法在离线环境中加载测试。这是设计使然：agent.py在模块顶层就创建了AgentRegistry并调用get_mcp_toolset。运行该示例时必须保证：

网络可达 Agent Registry API；
默认凭证有效且具备访问权限；
PROJECT_ID、MCP_SERVER_NAME对应的资源真实存在。

单元测试覆盖

AgentRegistry的完整行为由 tests/unittests/integrations/agent_registry/test_agent_registry.py 覆盖，主要包括：MCP Toolset 的 destination ID 注入、认证头注入策略、bindings认证方案解析、远程 A2A Agent 的构建、mTLS 端点选择、以及各类 API 错误处理，可作为你集成 Agent Registry 时的行为参考。

总结

api_registry_agent示例虽然代码简短，却完整展示了 ADK 集成 Google Cloud Agent Registry 的标准姿势：安装带a2a扩展的 ADK → 配置项目 ID / Location / MCP Server 名 → 用get_mcp_toolset一键解析连接与认证 → 注入LlmAgent成为可用工具。在其背后，agent_registry.py 承担了 endpoint 解析、认证绑定自动发现、Bearer Token 注入、mTLS 支持以及遥测标识注入等全部底层工作，让开发者可以专注于 Agent 的提示词与业务逻辑本身。这套模式同样适用于 A2A 远程 Agent 的发现与调用，是构建"服务目录驱动型"多 Agent 应用的实用基础。

创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考

使用 ADK 的 AgentRegistry 集成 Google Cloud 服务：以 BigQuery MCP 工具为例的完整实战指南