Google 会使用 AI 技术将内容翻译成您偏好的语言。AI 翻译可能包含错误。

使用 Agent Registry 和 ADK 动态发现 MCP 服务器

1. 简介

在本 Codelab 中，您将学习如何使用 Google Cloud Agent Registry 和智能体开发套件 (ADK) 在运行时动态发现和调用已注册的 Model Context Protocol (MCP) 服务器，从而消除硬编码端点和工具配置。

本实验使用 ADK 和 Python 编程语言来构建应用。这些原则同样适用于其他编程语言和智能体开发框架。

您将执行的操作

在项目中启用并配置 Google Cloud Agent Registry
将 MCP 服务器部署到 Cloud Run，并在 Agent Registry 中注册
构建一个使用硬编码网址调用 MCP 服务器的 ADK 智能体
更改智能体，使其通过身份在 Agent Registry 中发现 MCP 服务器
将智能体更改为在运行时使用语义搜索发现 MCP 服务器

所需条件

新版网络浏览器（例如 Chrome）
启用了结算功能的 Google Cloud 项目
拥有在 Google Cloud 云项目中创建和管理资源以及配置 IAM 政策的权限的账号

2. 设置您的环境

本 Codelab 中的说明假定您在 Google Cloud 控制台的 Cloud Shell 中运行命令。Cloud Shell 是在 Google Cloud 中运行的一个命令行环境，其中预加载了必要的工具。如果您在本地环境中安装了 gcloud CLI、git 和 Python 3，也可以在本地环境中运行这些命令。

创建 Google Cloud 项目

在 Google Cloud 控制台的项目选择器页面上，创建 Google Cloud 项目。
确保您的 Cloud 项目已启用结算功能。了解如何检查项目是否已启用结算功能。

启动 Cloud Shell 终端

使用资源选择器选择您创建的 Google Cloud 项目。
点击 Google Cloud 控制台顶部的激活 Cloud Shell 。
连接到 Cloud Shell 后，验证您的身份：
```
gcloud auth list
```
确认您的项目已配置：
```
gcloud config get project
```
如果您的项目未按预期设置，或者您使用的是本地机器上的环境而非 Cloud Shell，请设置项目 ID：
```
export GOOGLE_CLOUD_PROJECT=<YOUR_PROJECT_ID>
gcloud config set project "${GOOGLE_CLOUD_PROJECT}"
```

选择 Google Cloud 项目并设置环境变量

启用所需的 Google Cloud API：

gcloud services enable \
  agentregistry.googleapis.com \
  aiplatform.googleapis.com \
  cloudbuild.googleapis.com \
  run.googleapis.com \
  --project="${GOOGLE_CLOUD_PROJECT}"

设置 Python 环境

创建并激活虚拟环境，然后安装所需的依赖项：

mkdir -p adk-dynamic-discovery && cd adk-dynamic-discovery
python3 -m venv .venv
source .venv/bin/activate
pip install "fastmcp<5.0.0" "google-adk[mcp,a2a,agent-identity]>=2.5.0,<3.0.0"

3. 创建并注册 Zoo MCP 服务器

在此步骤中，您将创建并注册一个 Zoo MCP 服务器。该服务器公开了三个工具：list_species - 列出动物园中的所有物种；get_animals_by_species - 列出特定物种的动物；get_animal_details - 获取特定动物的其他信息。

您可以在 GitHub 上找到此步骤的所有文件。

创建 MCP 服务器

为 MCP 服务器设置文件夹和 Python 项目

mkdir mcp-on-cloudrun && cd mcp-on-cloudrun
uv init --description "Example of deploying an MCP server on Cloud Run" --bare --python 3.13
uv add "fastmcp<5.0.0" --no-sync

下载 MCP 服务器代码：

curl -fsSL https://raw.githubusercontent.com/GoogleCloudPlatform/devrel-demos/main/codelabs/agent-registry-dynamic-discovery-with-adk/steps/step3/server.py -o server.py

将 MCP 服务器部署到 Cloud Run，并在 Agent Registry 中注册

接下来，您将为服务器创建 Dockerfile，并将服务器部署到 Cloud Run。

创建 Dockerfile：

cat << EOF > Dockerfile
# Use the official Python image
FROM python:3.13-slim

# Install uv
COPY --from=ghcr.io/astral-sh/uv:latest /uv /uvx /bin/

# Install the project into /app
COPY . /app
WORKDIR /app

# Allow statements and log messages to immediately appear in the logs
ENV PYTHONUNBUFFERED=1

# Install dependencies
RUN uv sync

# Run the FastMCP server
CMD ["uv", "run", "server.py"]
EOF

为服务器创建指定的服务账号：

gcloud iam service-accounts create zoo-mcp-server-sa \
    --display-name="Zoo MCP Server Service Account"

将服务器部署到 us-west1 区域。

gcloud run deploy zoo-mcp-server \
    --service-account=zoo-mcp-server-sa@$GOOGLE_CLOUD_PROJECT.iam.gserviceaccount.com \
    --region=us-west1 \
    --source=. \
    --labels=dev-tutorial=codelab-dynamic-mcp-discovery \
    --invoker-iam-check \
    --no-allow-unauthenticated

如果您看到以下提示，请确认创建新的 Artifact Registry 代码库。如果您的项目中没有 us-west1 区域的 Artifact Registry 代码库，系统会显示以下提示：

Deploying from source requires an Artifact Registry Docker repository to store built containers. A repository named
[cloud-run-source-deploy] in region [us-west1] will be created.

Do you want to continue (Y/n)?

在 Agent Registry 的“global”位置注册 MCP 服务器：

fastmcp inspect server.py --format mcp | jq 'del(..|nulls) | del(.tools[]?._meta?)' > toolspec.json
MCP_URL=$(gcloud run services describe zoo-mcp-server --region=us-west1 --format='value(status.url)')
gcloud agent-registry services create zoo-mcp-server \
    --display-name="zoo-mcp-server" \
    --project="${GOOGLE_CLOUD_PROJECT}" \
    --location=global \
    --mcp-server-spec-type="tool-spec" \
    --mcp-server-spec-content="toolspec.json" \
    --interfaces="url=${MCP_URL}/mcp,protocolBinding=jsonrpc"

这些命令会生成包含工具清单的 toolspec.json，并调用 Agent Registry API 来注册服务器。

验证代理是否显示在注册表中：

gcloud agent-registry mcp-servers list \
    --location=global \
    --filter="display_name=zoo-mcp-server"

您将看到包含 MCP 服务器信息的 YAML 输出，如下所示：

---
attributes:
  agentregistry.googleapis.com/system/RuntimeReference:
    uri: //agentregistry.googleapis.com/projects/[PROJECT_NUMBER]/locations/global/services/zoo-mcp-server
createTime: '<create time datetime in ISO 8601>'
displayName: zoo-mcp-server
interfaces:
- protocolBinding: JSONRPC
  url: https://zoo-mcp-server-[PROJECT_NUMBER].us-west1.run.app/mcp
mcpServerId: urn:mcp:projects-[PROJECT_NUMBER]:projects:[PROJECT_NUMBER]:locations:global:agentregistry:services:zoo-mcp-server
name: projects/[PROJECT_ID]/locations/global/mcpServers/agentregistry-00000000-0000-0000-1234-56789abcdef
tools:
- annotations:
    idempotentHint: true
    readOnlyHint: true
  description: |-
    Returns every species in the zoo. Use it first for zoo-wide questions
    such as counting all animals or listing everything the zoo has.
  name: list_species
- annotations:
    idempotentHint: true
    readOnlyHint: true
  description: |-
    Retrieves all animals of a specific species from the zoo.
    Can also be used to collect the base data for aggregate queries
    of animals of a specific species - like counting the number of penguins
    or finding the oldest lion.
  name: get_animals_by_species
- annotations:
    idempotentHint: true
    readOnlyHint: true
  description: Retrieves the details of a specific animal by its name.
  name: get_animal_details
updateTime: '<last updated datetime in ISO 8601>'

4. 创建基本 ADK 智能体以使用 Zoo MCP 服务器

在此步骤中，您将创建一个 ADK 智能体，借助硬编码的 Zoo MCP 服务器网址来回答用户关于动物园的请求。硬编码会强制您在部署智能体之前知道 MCP 服务器的网址。

您可以在 GitHub 上找到此步骤的源文件。

创建智能体应用

创建 ADK 项目和使用 Zoo MCP 服务器的 Zoo 智能体的基架。

为 Zoo 代理设置文件夹和项目：

mkdir ../sample-agent && cd ../sample-agent
uv init --description "Example of ADK agent to showcase dynamic MCP discovery" --bare --python 3.13
uv add "google-adk[mcp,a2a,agent-identity]>=2.5.0,<3.0.0" --no-sync

创建代理软件包文件夹并下载 __init__.py 文件：

mkdir zoo_agent
curl -fsSL https://raw.githubusercontent.com/GoogleCloudPlatform/devrel-demos/main/codelabs/agent-registry-dynamic-discovery-with-adk/steps/step4/zoo_agent/__init__.py -o zoo_agent/__init__.py

创建 .env 文件，并使用此步骤和后续步骤中使用的环境变量填充该文件：
```
MCP_URL=$(gcloud run services describe zoo-mcp-server --region=us-west1 --format='value(status.url)')
cat << EOF > zoo_agent/.env
GOOGLE_GENAI_USE_ENTERPRISE=TRUE
GOOGLE_CLOUD_LOCATION=global
GOOGLE_CLOUD_PROJECT="${GOOGLE_CLOUD_PROJECT}"
GOOGLE_CLOUD_PROJECT_NUMBER=$(gcloud projects describe ${GOOGLE_CLOUD_PROJECT} \
   --format='value(projectNumber)')
MCP_SERVER_URL="${MCP_URL}"
EOF
```
此命令会将此步骤和后续发现步骤所需的所有环境变量写入 zoo_agent/.env，因此您以后无需重新创建该文件。在此步骤中，将使用以下变量：
- GOOGLE_GENAI_USE_ENTERPRISE、GOOGLE_CLOUD_PROJECT 和 GOOGLE_CLOUD_LOCATION：由 ADK 和 Google GenAI SDK 隐式使用，用于通过 Google Cloud 项目和位置 (global) 中的 Vertex AI 路由 Gemini 模型调用。它还存储 MCP 服务器注册到的 Agent Registry 的位置。
- GOOGLE_CLOUD_PROJECT_NUMBER：此步骤中不使用；此处预先填充，供下一步使用。
- MCP_SERVER_URL：直接在 zoo_agent/agent.py 中读取，以使用 Zoo MCP 服务器的硬编码 Cloud Run 网址配置 McpToolset。

下载 agent.py，其中包含用于连接到动物园 MCP 服务器的动物园智能体的代码：

curl -fsSL https://raw.githubusercontent.com/GoogleCloudPlatform/devrel-demos/main/codelabs/agent-registry-dynamic-discovery-with-adk/steps/step4/zoo_agent/agent.py -o zoo_agent/agent.py

MCP 服务器授权的工作原理

由于 zoo-mcp-server Cloud Run 服务是使用 --invoker-iam-check 和 --no-allow-unauthenticated 标志部署的，因此向 MCP 服务器发出的每个 HTTP 请求都必须在 Authorization 标头中包含 Google 签名的 ID 令牌。该令牌必须标识对服务具有 run.routes.invoke 权限的调用方。

在 zoo_agent/agent.py 中，_GoogleAuthHeaders 类通过以下两种方式之一获取 ID 令牌凭证。首先，它会调用 fetch_id_token_credentials()，从元数据服务器或服务账号密钥中获取服务网址的 ID 令牌。如果此操作失败，则会使用用户应用默认凭证 (ADC) 的 ID 令牌。每次调用该实例时，它都会根据需要刷新令牌并返回 Authorization 标头：

class _GoogleAuthHeaders:
    def __init__(self, audience: str):
        self._request = Request()
        self._id_token = False
        try:
            self._credentials = id_token.fetch_id_token_credentials(audience, request=self._request)
        except exceptions.DefaultCredentialsError:
            self._credentials, _ = google.auth.default()
            self._id_token = True
            if not isinstance(self._credentials, credentials.Credentials):
                raise

    def __call__(self, unused_context: ReadonlyContext | None = None) -> dict[str, str]:
        if not self._credentials.valid or (self._id_token and not self._credentials.id_token):
            self._credentials.refresh(self._request)
        token = self._credentials.id_token if self._id_token else self._credentials.token
        return {"Authorization": f"Bearer {token}"}

智能体会将该类的实例作为 header_provider 参数传递给 McpToolset。ADK 会调用标头提供程序，并将返回的标头附加到发送至 MCP 服务器的每个请求中。由于 HTTP 客户端是由 ADK 创建的，因此无论安装了哪个版本的 MCP Python SDK 及其 HTTP 库，相同的代码都可以正常运行：

zoo_mcp_toolset = McpToolset(
    connection_params=StreamableHTTPConnectionParams(
        url=f"{MCP_SERVER_URL}/mcp",
        timeout=30.0,
    ),
    header_provider=_GoogleAuthHeaders(MCP_SERVER_URL),
)

请求的受众群体是服务网址 (MCP_SERVER_URL)，而连接网址包含 FastMCP 服务器提供 MCP 协议的 /mcp 路径。30 秒的超时时间为 Cloud Run 冷启动留出了空间。

在本地环境中使用 ADK Playground 测试智能体

在测试智能体之前，请确保您的用户账号可以调用 MCP 服务器。

仅限本地终端（在 Cloud Shell 中跳过此步骤）：使用您的用户凭据初始化 ADC：
```
gcloud auth application-default login
```
按照说明完成 ADC 初始化。
（可选）向您的用户账号授予调用远程 MCP 服务器的权限：
```
gcloud run services add-iam-policy-binding zoo-mcp-server \
   --region=us-west1 \
   --project=${GOOGLE_CLOUD_PROJECT} \
   --member=user:$(gcloud config get-value account) \
   --role='roles/run.invoker'
```
该命令会遵循最小权限原则，仅向用户授予调用 MCP 服务器的权限。如果您是 Project Owner，则您已经拥有此权限；对于角色范围较窄的用户，则需要进行绑定。
使用 adk run 命令运行智能体：
```
adk run zoo_agent
```
或使用 adk web 命令运行 ADK 网页界面：
```
adk web --port 8080
```
然后在浏览器中打开该命令输出的网址。如果您使用 Cloud Shell，请添加 --allow_origins 参数，以允许从 Cloud Shell 打开网页界面：
```
adk web --port 8080 --allow_origins="regex:https://.*\.cloudshell\.dev"
```
运行该命令后，在 Cloud Shell 工具栏（右侧）中找到网页预览 图标。点击该图标，然后选择在端口 8080 上预览。Cloud Shell 将打开一个新的浏览器标签页，其中显示 ADK 网页界面。
开始与代理聊天。例如，询问动物园里的动物：
```
How many animals are in the zoo?
```
智能体针对每个物种调用 list_species，然后调用 get_animals_by_species，以计算出答案。

通过 ADK 网页界面，您可以探索智能体的工作原理，包括它对动物园 MCP 服务器的调用。如需了解具体方法，请参阅网页界面文档。

5. 动态发现 MCP 服务器端点

在此步骤中，您将更改智能体，使其发现 Agent Registry 中注册的动物园 MCP 服务器端点，而不是对其端点网址进行硬编码。

您可以在 GitHub 上找到此步骤的所有文件。

您可以使用 ADK 与 Agent Registry 的集成来查找和检索服务器的网址。如需实现此功能，您需要以下信息：

MCP 服务器注册到的 Agent Registry 的项目编号和位置 (global)。
您用于注册 MCP 服务器的服务 ID (zoo-mcp-server)。

将智能体更新为使用 Agent Registry

从 GitHub 下载更新后的 agent.py 文件，以覆盖第一个版本的代理：
```
curl -fsSL https://raw.githubusercontent.com/GoogleCloudPlatform/devrel-demos/main/codelabs/agent-registry-dynamic-discovery-with-adk/steps/step5/zoo_agent/agent.py -o zoo_agent/agent.py
```
此步骤中的环境变量
此次更新后，代理将依赖于 zoo_agent/.env 中的以下环境变量：
- GOOGLE_CLOUD_PROJECT_NUMBER：现在，构建 MCP 服务器的身份 URN (MCP_SERVER_URN) 时需要提供此参数。
- GOOGLE_CLOUD_PROJECT、GOOGLE_GENAI_USE_ENTERPRISE 和 GOOGLE_CLOUD_LOCATION：由 ADK 用于配置 Vertex AI 访问权限。该位置还用于发现 MCP 服务器。
- 不再需要：MCP_SERVER_URL 不再使用，因为智能体会从 Agent Registry 检索 MCP 服务器端点网址，而不是从环境中读取。
MCP 服务器身份和发现在 AgentRegistry 中的工作原理
在智能体注册表中，每个已注册的 MCP 服务器都有一个存储在其 mcpServerId 字段中的唯一标识符，格式为统一资源名称 (URN)。您可以在第 3 步中 gcloud agent-registry mcp-servers list 命令的输出中看到该值。代理会根据以下环境变量重构此标识符：
```
PROJECT_ID = os.environ["GOOGLE_CLOUD_PROJECT"]
PROJECT_NUMBER = os.environ["GOOGLE_CLOUD_PROJECT_NUMBER"]
LOCATION = os.environ.get("GOOGLE_CLOUD_LOCATION", "global")
MCP_SERVER_URN = f"urn:mcp:projects-{PROJECT_NUMBER}:projects:{PROJECT_NUMBER}:locations:{LOCATION}:agentregistry:services:zoo-mcp-server"
```
URN 会对注册表的项目编号 (projects-{PROJECT_NUMBER})、服务器注册所在的项目编号和位置 (projects:{PROJECT_NUMBER}:locations:{LOCATION}) 以及您在 gcloud agent-registry services create 命令中使用的服务 ID (agentregistry:services:zoo-mcp-server) 进行编码。智能体会使用 registry.list_mcp_servers() 列出已注册的 MCP 服务器，并将每个服务器的 mcpServerId 与 MCP_SERVER_URN 进行匹配。然后，它会将匹配的服务器资源 name (projects/.../locations/.../mcpServers/...) 传递给 registry.get_mcp_toolset()，后者会从服务器的注册接口 (https://...run.app/mcp) 读取端点网址，并返回连接到该网址的 McpToolset：
```
header_provider = _GoogleAuthHeaders()
registry = AgentRegistry(
    project_id=PROJECT_ID,
    location=LOCATION,
    header_provider=header_provider,
)

mcp_server_name = next(
    (
        s["name"]
        for s in registry.list_mcp_servers().get("mcpServers", [])
        if s.get("mcpServerId") == MCP_SERVER_URN
    ),
    None,
)
if not mcp_server_name:
    raise RuntimeError(f"MCP server '{MCP_SERVER_URN}' not found in Agent Registry.")
zoo_mcp_toolset = registry.get_mcp_toolset(mcp_server_name=mcp_server_name)
header_provider.bind(zoo_mcp_toolset.connection_params.url)
zoo_mcp_toolset.connection_params.timeout = 30.0
```
get_mcp_toolset() 会在工具名称前加上服务器的显示名称，因此智能体看到的工具名称为 zoo_mcp_server_list_species、zoo_mcp_server_get_animals_by_species 和 zoo_mcp_server_get_animal_details。当智能体使用来自多个 MCP 服务器的工具时，前缀可防止名称冲突。最后一行将连接超时从默认的 5 秒提高，以便为 Cloud Run 冷启动留出空间。
AgentRegistry 如何改变授权
AgentRegistry 使用 ADC 对其自己的 API 调用（例如 list_mcp_servers()）进行身份验证，因此只有在调用发现的 MCP 服务器时才需要标头提供程序。由于 AgentRegistry 会在内部创建 McpToolset，因此您需要将标头提供程序传递给 AgentRegistry 构造函数。在上一步中，受众群体是您事先知道的 MCP_SERVER_URL 值。现在，只有在发现之后，网址才会被知晓。ADK 会在智能体向 MCP 服务器发送请求时调用标头提供程序，而不是在创建 AgentRegistry 时调用。这样，代理就可以先创建提供程序，并在 get_mcp_toolset() 返回后将其绑定到发现的端点。bind() 方法获取凭据的方式与上一步中的构造函数相同，受众群体取自发现的端点网址，不含 /mcp 路径：
```
    def bind(self, endpoint_url: str) -> None:
        parsed = urlparse(endpoint_url)
        audience = f"{parsed.scheme}://{parsed.netloc}"
        try:
            self._credentials = id_token.fetch_id_token_credentials(audience, request=self._request)
        except exceptions.DefaultCredentialsError:
            self._credentials, _ = google.auth.default()
            self._id_token = True
            if not isinstance(self._credentials, credentials.Credentials):
                raise
```
__call__ 方法与上一步中的方法相同，但如果在 bind() 之前调用，则会引发错误。
运行智能体以测试更改。使用 adk run 命令：
```
adk run zoo_agent
```
或使用 adk web 命令运行 ADK 网页界面：
```
adk web --port 8080
```
然后在浏览器中打开该命令输出的网址。如果您使用 Cloud Shell，请添加 --allow_origins 参数，以允许从 Cloud Shell 打开网页界面：
```
adk web --port 8080 --allow_origins="regex:https://.*\.cloudshell\.dev"
```
向智能体提出与上一步相同的问题。回答相同，但在 ADK 网页界面中，工具调用现在显示为 zoo_mcp_server_list_species 和 zoo_mcp_server_get_animals_by_species。
问题排查：智能体在加载智能体模块时会发现 MCP 服务器。如果智能体启动失败并显示错误 MCP server '...' not found in Agent Registry，请验证 zoo_agent/.env 中 GOOGLE_CLOUD_PROJECT_NUMBER 和 GOOGLE_CLOUD_LOCATION 的值是否与您在第 3 步中注册服务器的项目和位置一致。

此项变更让您无需再提供明确的 MCP 服务器网址。不过，您仍需知道项目编号、注册表位置和用于注册 MCP 服务器的服务 ID，并且发现过程是在智能体初始化期间静态执行的。

6. 在动态发现中使用 MCP 服务器的语义搜索

在此步骤中，您将通过在启动时将精确 URN 匹配替换为运行时语义搜索，来改进 MCP 服务器动态发现。这种方法无需知道服务器部署所在的项目编号或服务器的确切身份 (MCP_SERVER_URN)，并将工具发现直接集成到智能体的标准运行时工作流中，而不是在初始化时处理。

您可以在 GitHub 上找到此步骤的所有文件。

将智能体更新为在运行时使用语义发现

从 GitHub 下载更新后的 agent.py 文件，以覆盖现有实现：

curl -fsSL https://raw.githubusercontent.com/GoogleCloudPlatform/devrel-demos/main/codelabs/agent-registry-dynamic-discovery-with-adk/steps/step6/zoo_agent/agent.py -o zoo_agent/agent.py

此步骤中的环境变量

由于智能体不再构建 MCP 服务器 URN，因此所需的配置进一步简化：

GOOGLE_CLOUD_PROJECT、GOOGLE_GENAI_USE_ENTERPRISE 和 GOOGLE_CLOUD_LOCATION：由 ADK 用于配置 Vertex AI 访问权限。该位置还用于发现 MCP 服务器。
不再需要：不再需要 GOOGLE_CLOUD_PROJECT_NUMBER（以及在上一步中已移除的 MCP_SERVER_URL）。

从完全匹配转向语义搜索

agent.py 现在定义了一个发现工具函数 discover_zoo_mcp_servers，该函数使用 search_type="SEMANTIC" 调用 registry.search_mcp_servers()，而不是在模块初始化期间调用 registry.list_mcp_servers() 并将每个条目与构造的 MCP_SERVER_URN 进行比较：

def discover_zoo_mcp_servers(query: str, tool_context: ToolContext) -> dict[str, Any]:
    """Discovers Zoo MCP server(s) matching the query using semantic search.

    Args:
        query: Semantic description of zoo capabilities needed (e.g. 'zoo animal info')
    """
    cached = _fresh_discovery(tool_context.state)
    if cached:
        return {
            "status": "already_discovered",
            "message": "Zoo MCP servers are already discovered.",
            "servers": cached["servers"],
        }

    try:
        response = registry.search_mcp_servers(
            search_string=query,
            search_type="SEMANTIC",
            page_size=5,
        )
    except Exception as e:
        return {
            "status": "error",
            "message": f"Agent Registry search failed: {e}",
        }

    found_servers = response.get("mcpServers", [])
    if not found_servers:
        return {
            "status": "not_found",
            "message": f"No MCP servers found in Agent Registry at {LOCATION} matching '{query}'.",
        }

    valid_servers = [s["name"] for s in found_servers if "name" in s]
    tool_context.state[DISCOVERY_STATE_KEY] = {
        "servers": valid_servers,
        "discovered_at": time.time(),
    }
    return {
        "status": "discovered",
        "message": f"Successfully discovered {len(valid_servers)} Zoo MCP server(s).",
        "servers": valid_servers,
    }

这样，智能体就可以在运行时使用所需功能的自然语言描述（例如 "zoo animals, exhibits, habitat, care, feeding schedule"）搜索 Agent Registry，并将发现的服务器资源名称保存在 ADK 会话状态 (tool_context.state) 中。

如何将动态发现集成到智能体工作流中

为了从运行时发现的服务器挂载工具，并引导智能体完成“发现-调用”流程，agent.py 将自定义 BaseToolset 实现 (RegistryDiscoveredToolset) 与动态指令提供程序 (zoo_guide_instruction) 结合使用：

动态工具集 (RegistryDiscoveredToolset)：子类 BaseToolset 并设置 self._use_invocation_cache = False，以便 ADK 在 discover_zoo_mcp_servers 更新会话状态后立即重新评估可用工具，而不是缓存最初为空的工具列表。调用 get_tools() 时，它会从状态中读取发现的服务器名称，在工作器线程中为每个服务器构建经过身份验证的 McpToolset，并返回所有活跃工具。无法连接到其中一个服务器不会影响其他服务器。_build_toolset() 会执行阻塞 I/O，因此 _tools_of() 会在工作器线程中运行它。

class RegistryDiscoveredToolset(BaseToolset):
    def __init__(self, agent_registry: AgentRegistry):
        super().__init__()
        self._registry = agent_registry
        self._toolsets: dict[str, McpToolset] = {}
        self._use_invocation_cache = False

    def _build_toolset(self, server_name: str) -> McpToolset:
        header_provider = _GoogleAuthHeaders()
        server_registry = AgentRegistry(
            project_id=self._registry.project_id,
            location=self._registry.location,
            header_provider=header_provider,
        )
        toolset = server_registry.get_mcp_toolset(server_name)
        header_provider.bind(toolset.connection_params.url)
        toolset.connection_params.timeout = 30.0
        return toolset

    async def _tools_of(self, server_name: str, ctx: ReadonlyContext) -> list[BaseTool]:
        try:
            if server_name not in self._toolsets:
                self._toolsets[server_name] = await asyncio.to_thread(
                    self._build_toolset, server_name
                )
            return await self._toolsets[server_name].get_tools_with_prefix(ctx)
        except Exception as e:
            print(f"Skipping MCP server {server_name}: {e}")
            return []

    async def get_tools(self, readonly_context: ReadonlyContext | None = None) -> list[BaseTool]:
        if not readonly_context:
            return []
        record = _fresh_discovery(readonly_context.state)
        if not record or not record.get("servers"):
            if self._toolsets:
                await self.close()
            return []
        per_server = await asyncio.gather(
            *(self._tools_of(name, readonly_context) for name in record["servers"])
        )
        return [tool for tools in per_server for tool in tools]

动态智能体指令 (zoo_guide_instruction)：在每个模型轮次之前评估会话状态。如果尚未发现任何服务器，它会附加 _DISCOVERY_PROMPT，指示代理先调用 discover_zoo_mcp_servers。发现成功后，它会更新提示，列出活跃的 MCP 服务器，并指示智能体直接使用已挂载的动物园工具，而无需再次运行发现：

def zoo_guide_instruction(ctx: ReadonlyContext) -> str:
    record = _fresh_discovery(ctx.state)
    if record is None:
        return _ROLE + _DISCOVERY_PROMPT
    return _ROLE + (
        f"\nYour active Zoo MCP servers: {', '.join(record['servers'])}. "
        "Do not run discovery again; use the active zoo tools directly to fulfill the visitor's request."
    )

动态发现的服务器的授权

语义搜索可能会返回多个服务器，而 ADK 会调用标头提供程序，但不会告知其请求会发送到哪个服务器。因此，_build_toolset() 会为每个发现的服务器创建一个专用的 _GoogleAuthHeaders 和 AgentRegistry，并将提供程序绑定到该服务器的端点。提供程序与上一步中的提供程序相同，但有一个 _bound 标志，如果该提供程序在 bind() 之前使用，则会导致失败。

保留和刷新逻辑

为了避免在每个回合都查询智能体注册表，同时仍能定期刷新发现的端点，agent.py 实现了具有存留时间 (TTL) 的状态保留：

DISCOVERY_STATE_KEY = "app:zoo_mcp_servers"
DISCOVERY_TTL_SECONDS = 24 * 60 * 60

def _fresh_discovery(state: Any) -> dict[str, Any] | None:
    if not state:
        return None
    record = state.get(DISCOVERY_STATE_KEY)
    if not record or not record.get("servers"):
        return None
    if time.time() - record.get("discovered_at", 0) > DISCOVERY_TTL_SECONDS:
        return None
    return record

应用范围的保留：发现的服务器名称和 discovered_at 时间戳存储在 app:zoo_mcp_servers 键下的会话状态中。使用 ADK 的 app: 前缀可在应用运行期间限定缓存的发现记录的范围。
TTL 验证和自动刷新：_fresh_discovery() 会验证保存的记录是否存在，以及其存在时间是否未超过 DISCOVERY_TTL_SECONDS（24 小时）。当记录处于最新状态时，RegistryDiscoveredToolset 会重复使用缓存的 McpToolset 连接 (self._toolsets)，zoo_guide_instruction 会省略发现提示，并且对 discover_zoo_mcp_servers 的任何重复调用都会立即返回 "already_discovered"。24 小时 TTL 到期后，_fresh_discovery() 会返回 None，从而导致 RegistryDiscoveredToolset.get_tools() 关闭现有 MCP 客户端会话 (await self.close())，并导致 zoo_guide_instruction 提示智能体执行新的发现。

运行智能体，测试语义发现工作流。使用 adk run 命令：
```
adk run zoo_agent
```
或使用 adk web 命令运行 ADK 网页界面：
```
adk web --port 8080
```
然后在浏览器中打开该命令输出的网址。如果您使用 Cloud Shell，请添加 --allow_origins 参数，以允许从 Cloud Shell 打开网页界面：
```
adk web --port 8080 --allow_origins="regex:https://.*\.cloudshell\.dev"
```
向智能体询问有关动物园的问题（例如 How many animals are in the zoo?），观察它如何首先调用 discover_zoo_mcp_servers，动态装载从发现的 MCP 服务器中找到的工具，然后调用动物园工具（例如 zoo_mcp_server_list_species 和 zoo_mcp_server_get_animals_by_species）来回答您的问题。提出一个后续问题，例如 Which is the oldest lion?。智能体使用已装载的动物园工具进行回答，而不会再次调用 discover_zoo_mcp_servers，因为会话状态中的发现记录仍然有效。

7. 清理

选项 1：删除项目

如果您为此 Codelab 创建了一个专用项目，那么避免产生费用的最简单方法是删除该项目：

gcloud projects delete "${GOOGLE_CLOUD_PROJECT}" --quiet

选项 2：删除单个资源

如果您使用的是现有项目，并且想要保留该项目，请删除在此 Codelab 期间创建的各个资源：

从 Agent Registry 中删除 MCP 服务器注册。由于您是手动注册服务器，因此删除 Cloud Run 服务并不会移除注册：

gcloud agent-registry services delete zoo-mcp-server \
    --location=global \
    --project="${GOOGLE_CLOUD_PROJECT}" \
    --quiet

删除 Cloud Run MCP 服务器服务：

gcloud run services delete zoo-mcp-server \
    --region=us-west1 \
    --project="${GOOGLE_CLOUD_PROJECT}" \
    --quiet

删除为 MCP 服务器创建的服务账号：

gcloud iam service-accounts delete \
    "zoo-mcp-server-sa@${GOOGLE_CLOUD_PROJECT}.iam.gserviceaccount.com" \
    --project="${GOOGLE_CLOUD_PROJECT}" \
    --quiet

如果 cloud-run-source-deploy Artifact Registry 制品库是在此 Codelab 期间专门创建的，请将其删除，以免产生容器映像存储费用：

gcloud artifacts repositories delete cloud-run-source-deploy \
    --location=us-west1 \
    --project="${GOOGLE_CLOUD_PROJECT}" \
    --quiet

（可选）移除在 Cloud Shell 中创建的本地项目目录：
```
cd ~ && rm -rf adk-dynamic-discovery
```

8. 恭喜

在此 Codelab 中，您学习了如何通过 Google Cloud Agent Registry 和智能体开发套件 (ADK) 动态发现和使用 MCP 服务器：

注册了 MCP 服务器：将安全的 MCP 服务器部署到 Cloud Run，并在 Agent Registry 中进行了注册。
通过身份发现 MCP 服务器：从 Agent Registry 解析 MCP 服务器端点，而不是对其网址进行硬编码。
通过语义搜索发现 MCP 服务器：让智能体在运行时使用自然语言搜索来查找和装载 MCP 工具，并支持保留和刷新。

使用 Agent Registry 和 ADK 动态发现 MCP 服务器

1. 简介

您将执行的操作

所需条件

2. 设置您的环境

创建 Google Cloud 项目

启动 Cloud Shell 终端

选择 Google Cloud 项目并设置环境变量

设置 Python 环境

3. 创建并注册 Zoo MCP 服务器

创建 MCP 服务器

将 MCP 服务器部署到 Cloud Run，并在 Agent Registry 中注册

4. 创建基本 ADK 智能体以使用 Zoo MCP 服务器

创建智能体应用

MCP 服务器授权的工作原理

在本地环境中使用 ADK Playground 测试智能体

5. 动态发现 MCP 服务器端点

将智能体更新为使用 Agent Registry

此步骤中的环境变量

MCP 服务器身份和发现在 `AgentRegistry` 中的工作原理

`AgentRegistry` 如何改变授权

6. 在动态发现中使用 MCP 服务器的语义搜索

将智能体更新为在运行时使用语义发现

此步骤中的环境变量

从完全匹配转向语义搜索

如何将动态发现集成到智能体工作流中

动态发现的服务器的授权

保留和刷新逻辑

7. 清理

选项 1：删除项目

选项 2：删除单个资源

8. 恭喜

后续步骤

使用 Agent Registry 和 ADK 动态发现 MCP 服务器

1. 简介

您将执行的操作

所需条件

2. 设置您的环境

创建 Google Cloud 项目

启动 Cloud Shell 终端

选择 Google Cloud 项目并设置环境变量

设置 Python 环境

3. 创建并注册 Zoo MCP 服务器

创建 MCP 服务器

将 MCP 服务器部署到 Cloud Run，并在 Agent Registry 中注册

4. 创建基本 ADK 智能体以使用 Zoo MCP 服务器

创建智能体应用

MCP 服务器授权的工作原理

在本地环境中使用 ADK Playground 测试智能体

5. 动态发现 MCP 服务器端点

将智能体更新为使用 Agent Registry

此步骤中的环境变量

MCP 服务器身份和发现在 AgentRegistry 中的工作原理

AgentRegistry 如何改变授权

6. 在动态发现中使用 MCP 服务器的语义搜索

将智能体更新为在运行时使用语义发现

此步骤中的环境变量

从完全匹配转向语义搜索

如何将动态发现集成到智能体工作流中

动态发现的服务器的授权

保留和刷新逻辑

7. 清理

选项 1：删除项目

选项 2：删除单个资源

8. 恭喜

后续步骤

MCP 服务器身份和发现在 `AgentRegistry` 中的工作原理

`AgentRegistry` 如何改变授权