向工具和资源进行身份验证

借助 Agent Registry，您的编排器智能体可以安全地向发现的远程智能体、Model Context Protocol (MCP) 工具集和Google Cloud 服务进行身份验证。使用 Agent Identity 和 Agent Identity 身份验证管理器，您可以在调用通过 Agent Registry 发现的功能时配置凭证、绑定和自定义 HTTP 标头。

准备工作

在向工具和资源进行身份验证之前，请完成以下操作：

1. 设置 Agent Registry。

2. 获取已启用 Agent Registry 的治理项目的项目 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
   

   ADC 凭证必须具有代理或工具与之交互的底层 Google Cloud 服务所需的 Identity and Access Management (IAM) 权限。

5. 设置以下环境变量：

   export GOOGLE_CLOUD_PROJECT=PROJECT_ID
   export GOOGLE_CLOUD_LOCATION=LOCATION
   export API_KEY=EXTERNAL_API_KEY
   

   替换以下内容：

   • PROJECT_ID：治理项目的项目 ID。
   • LOCATION：注册表区域或位置，例如 us-central1。
   • EXTERNAL_API_KEY：如果使用自定义标头，则为您的外部 API 密钥。

了解身份验证模型

为了管理凭证，Agent Registry 依赖 Agent Identity 进行内置Google Cloud 身份验证，并依赖身份验证管理器来管理身份验证提供方和绑定。

您的智能体可以使用自己的身份，也可以使用身份验证管理器通过 API 密钥或 OAuth 2.0 访问工具。无论您使用的是代理身份还是服务账号，代理自身的身份都是通过应用默认凭证 (ADC) 提供的。

如果您需要在不使用身份验证管理器的情况下进行身份验证，请直接在 ADK 请求中提供自定义 HTTP 标头。

如需确定最适合您的使用场景的方法，请参阅身份验证模型。

为 Google Cloud 工具使用智能体身份

对于 Google Cloud 服务（例如 Gemini Enterprise Agent Platform 上的 Agent Runtime 或 Cloud Storage），智能体会使用自己的身份访问工具。无论您使用的是代理身份还是服务账号，此访问权限都通过应用默认凭证 (ADC) 进行管理。

与您的代理关联的身份必须具有底层 Google Cloud 服务所需的 IAM 权限。例如，如果 MCP 工具管理 Compute Engine 虚拟机实例，则除了 Agent Registry API Viewer 角色之外，智能体身份还必须具有 Compute Instance Admin (v1) 或更高级别等角色。

对远程 A2A 智能体进行身份验证

如果您的编排器智能体使用自己的身份连接到远程 Google Agent2Agent (A2A) 智能体，则必须在 get_remote_a2a_agent() 方法中明确包含使用 Google 身份验证标头配置的 httpx.AsyncClient。

为防止超出默认 HTTP 客户端超时限制，请定义自定义超时。

以下示例展示了如何使用 ADC 和自定义超时配置 httpx.AsyncClient：

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

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

# Initialize the registry client
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,
)

# Configure the HTTP client with GoogleAuth and a 60-second timeout
httpx_client = httpx.AsyncClient(auth=GoogleAuth(), timeout=httpx.Timeout(60.0))

# 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"
my_remote_agent = registry.get_remote_a2a_agent(
    agent_name=agent_name,
    httpx_client=httpx_client,
)

将 AGENT_ID 替换为注册代理的 ID。

对 MCP 工具进行身份验证

如果您的编排器智能体使用自己的身份访问 MCP 工具集，您可以使用 ADC 和 get_mcp_toolset() 方法初始化注册表客户端：

import os
from google.adk.integrations.agent_registry import AgentRegistry

# Initialize the registry client
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,
)

# 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)

将 SERVER_ID 替换为您注册的 MCP 服务器的 ID。

使用身份验证管理器进行自定义工具和委托访问

当您的智能体需要连接到外部 API、自定义工具或外部 MCP 服务器时，您可以使用身份验证管理器来管理凭据，而无需在应用中对其进行硬编码。

根据您的使用场景，您可以针对以下场景配置身份验证管理器：

• 访问自定义工具：当代理需要使用自己的身份和权限访问工具时，使用 API 密钥或 2-legged OAuth (2LO) 对代理进行身份验证。
• 代表用户访问工具：使用三方模式 OAuth (3LO) 对充当单个用户的智能体进行身份验证。如果代理必须代表与其互动的用户访问工具，则建议使用此模型，这需要用户同意和委托权限。

这些模型要求您创建身份验证提供方。然后，您需要在 Agent Registry 中创建一个绑定，将身份验证提供方与您的注册表资源关联起来。

配置身份验证提供方后，ADK 会管理编排器的身份验证流程。如需查看示例，请参阅在 ADK 代码中解析绑定。

访问自定义工具

如果您希望智能体使用自己的凭证连接到外部或自定义工具，可以将身份验证管理器配置为处理 API 密钥或双向 OAuth (2LO)：

1. 创建身份验证提供方：在身份验证管理器中创建身份验证提供方，以存储 API 密钥或 OAuth 凭证。如需了解相关信息，请参阅使用身份验证管理器通过两方模式 OAuth 进行身份验证或使用身份验证管理器通过 API 密钥进行身份验证。

2. 绑定身份验证提供方：为了让编排器代理检测目标的正确身份验证提供方，您必须在代理注册表中在代理、目标资源和身份验证提供方之间创建绑定。借助绑定，您无需在代码中手动定义身份验证提供方：

   gcloud

   通过运行 gcloud agent-registry bindings create 命令创建绑定，以将智能体明确关联到目标资源和身份验证提供方。指定 --auth-provider-binding 资源名称时，您必须使用治理项目的项目 ID：

   gcloud agent-registry bindings create BINDING_NAME \
       --project=PROJECT_ID \
       --location=LOCATION \
       --display-name="DISPLAY_NAME" \
       --source-identifier="SOURCE_ID" \
       --target-identifier="TARGET_ID" \
       --auth-provider-binding="projects/PROJECT_ID/locations/LOCATION/authProviders/AUTH_PROVIDER_ID"
   

   Terraform

   配置 google_agent_registry_binding 资源，将您的智能体和目标资源关联到身份验证提供方：

   resource "google_agent_registry_binding" "custom_tool_binding" {
     project      = "PROJECT_ID"
     location     = "LOCATION"
     binding_id   = "BINDING_NAME"
     display_name = "DISPLAY_NAME"
   
     source {
       identifier = "SOURCE_ID"
     }
   
     target {
       identifier = "TARGET_ID"
     }
   
     auth_provider_binding {
       auth_provider = "projects/PROJECT_ID/locations/LOCATION/authProviders/AUTH_PROVIDER_ID"
     }
   }
   

   替换以下内容：

   • BINDING_NAME：绑定的唯一 ID。
   • PROJECT_ID：治理项目的项目 ID。
   • LOCATION：注册表区域或位置，例如 us-central1 或 global。
   • DISPLAY_NAME：绑定的人类可读显示名称。
   • SOURCE_ID：URN 格式的源代理的代理标识符。
   • TARGET_ID：目标 MCP 服务器或端点的注册表标识符（采用 URN 格式）。
   • AUTH_PROVIDER_ID：身份验证提供方的 ID。

   如需详细了解如何管理这些连接，请参阅管理绑定。

代表用户访问工具

如果您希望智能体代表个人用户向远程 MCP 服务器或工具进行身份验证，请通过身份验证管理器使用三方模式 OAuth (3LO)。

授权管理器提供一项完全由 Google 代管式服务，用于处理 OAuth 令牌、用户同意和重定向。当最终用户与智能体互动并触发需要委托权限的工具时，平台会自动提示用户同意、存储凭证并恢复工作流：

1. 创建身份验证提供方：在身份验证管理器中创建身份验证提供方，并配置 OAuth 重定向 URI。如需了解相关信息，请参阅使用身份验证管理器通过三方模式 OAuth 进行身份验证。
2. 更新客户端应用：更新应用客户端，以处理 adk_request_credential 函数调用并管理用户同意情况。如需查看详细说明，请参阅更新客户端应用。

3. 绑定身份验证提供方：为了让编排器代理检测目标的正确身份验证提供方，您必须在代理注册表中在代理、目标资源和身份验证提供方之间创建绑定。借助绑定，您无需在代码中手动定义身份验证提供方：

   gcloud

   通过运行 gcloud agent-registry bindings create 命令创建绑定，以将智能体明确关联到目标资源和身份验证提供方。指定 --auth-provider-binding 资源名称时，您必须使用治理项目的项目 ID：

   gcloud agent-registry bindings create BINDING_NAME \
       --project=PROJECT_ID \
       --location=LOCATION \
       --display-name="DISPLAY_NAME" \
       --source-identifier="SOURCE_ID" \
       --target-identifier="TARGET_ID" \
       --auth-provider-binding="projects/PROJECT_ID/locations/LOCATION/authProviders/AUTH_PROVIDER_ID" \
       --auth-provider-binding-scopes=SCOPES \
       --auth-provider-binding-continue-uri="https://REDIRECT_URI"
   

   Terraform

   使用 OAuth 2.0 范围和继续 URI 配置 google_agent_registry_binding 资源：

   resource "google_agent_registry_binding" "user_delegated_binding" {
     project      = "PROJECT_ID"
     location     = "LOCATION"
     binding_id   = "BINDING_NAME"
     display_name = "DISPLAY_NAME"
   
     source {
       identifier = "SOURCE_ID"
     }
   
     target {
       identifier = "TARGET_ID"
     }
   
     auth_provider_binding {
       auth_provider = "projects/PROJECT_ID/locations/LOCATION/authProviders/AUTH_PROVIDER_ID"
       scopes        = [SCOPES]
       continue_uri  = "https://REDIRECT_URI"
     }
   }
   

   替换以下内容：

   • BINDING_NAME：绑定的唯一 ID。
   • PROJECT_ID：治理项目的项目 ID。
   • LOCATION：注册表区域或位置，例如 us-central1 或 global。
   • DISPLAY_NAME：绑定的人类可读显示名称。
   • SOURCE_ID：URN 格式的源代理的代理标识符。
   • TARGET_ID：目标 MCP 服务器或工具资源的注册表标识符，采用 URN 格式。
   • AUTH_PROVIDER_ID：身份验证提供方的 ID。
   • SCOPES：以英文逗号分隔的 OAuth 2.0 范围列表，例如 https://www.googleapis.com/auth/cloud-platform。在 Terraform 中，请用英文双引号括起每个范围。
   • REDIRECT_URI：用户同意后为身份验证提供方配置的重定向 URI。

   如需详细了解如何管理这些连接，请参阅管理绑定。

解析 ADK 代码中的绑定

在 Agent Registry 中创建身份验证提供方绑定后，ADK 会处理身份验证流程。

当您使用 AgentRegistry 客户端提取 MCP 工具集时，ADK 会自动解析与该服务器关联的所有绑定，并应用正确的架构。您无需在代码中手动配置凭据。

以下示例演示了如何提取经过身份验证的 MCP 工具集并将其附加到智能体：

import os
from google.adk.agents.llm_agent import LlmAgent
from google.adk.auth.credential_manager import CredentialManager
from google.adk.integrations.agent_identity import GcpAuthProvider
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.")

# Register the auth provider
CredentialManager.register_auth_provider(GcpAuthProvider())

# Initialize the registry client
registry = AgentRegistry(
    project_id=project_id,
    location=location,
)

# Fetch the registered MCP toolset.
# The ADK applies the bindings configured in Agent Registry.
mcp_server_name = "mcpServers/SERVER_ID"
my_mcp_toolset = registry.get_mcp_toolset(mcp_server_name=mcp_server_name)

# Compose the agent
maps_agent = LlmAgent(
    name="maps_agent",
    model="gemini-1.5-flash",
    instruction=(
        "You are a local guide and navigation expert. Your goal is to provide "
        "accurate location information and directions.\n\n"
        "Rules:\n"
        "1. **Planning**: Before answering, plan how to use the tools to get "
        "the best information (e.g., search for a place first, then get details "
        "or directions).\n"
        "2. **Tool Usage**: Use the Maps MCP tools to find locations, addresses, "
        "and navigation details. Do not guess locations or distances.\n"
        "3. **Output Format**: Provide clear summaries of locations, including "
        "full addresses and key details. Use formatting to make recommendations "
        "easy to read.\n"
        "4. **Tone**: Be helpful, enthusiastic, and descriptive, with "
        "appropriate emojis to enhance the experience."
    ),
    tools=[my_mcp_toolset],
)

将 SERVER_ID 替换为您注册的 MCP 服务器的 ID。

如果您设置了 3LO 来代表用户访问工具，则还必须向 get_mcp_toolset 方法提供 continue_uri。此 URI 指示最终用户在表示同意后被重定向到的位置：

import os
from google.adk.agents.llm_agent import LlmAgent
from google.adk.auth.credential_manager import CredentialManager
from google.adk.integrations.agent_identity import GcpAuthProvider
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.")

# Register the auth provider
CredentialManager.register_auth_provider(GcpAuthProvider())

# Initialize the registry client
registry = AgentRegistry(
    project_id=project_id,
    location=location,
)

# Fetch the registered MCP toolset.
# The ADK applies the bindings configured in Agent Registry.
# For 3-legged OAuth (3LO), provide continue_uri.
mcp_server_name = "mcpServers/SERVER_ID"
my_mcp_toolset = registry.get_mcp_toolset(
    mcp_server_name=mcp_server_name,
    # Replace with your app's redirect URI:
    continue_uri="https://REDIRECT_URI"
)

# Compose the agent
maps_agent = LlmAgent(
    name="maps_agent",
    model="gemini-1.5-flash",
    instruction=(
        "You are a local guide and navigation expert. Your goal is to provide "
        "accurate location information and directions.\n\n"
        "Rules:\n"
        "1. **Planning**: Before answering, plan how to use the tools to get "
        "the best information (e.g., search for a place first, then get details "
        "or directions).\n"
        "2. **Tool Usage**: Use the Maps MCP tools to find locations, addresses, "
        "and navigation details. Do not guess locations or distances.\n"
        "3. **Output Format**: Provide clear summaries of locations, including "
        "full addresses and key details. Use formatting to make recommendations "
        "easy to read.\n"
        "4. **Tone**: Be helpful, enthusiastic, and descriptive, with "
        "appropriate emojis to enhance the experience."
    ),
    tools=[my_mcp_toolset],
)

替换以下内容：

• SERVER_ID：已注册的 MCP 服务器的 ID。
• REDIRECT_URI：应用的重定向 URI，用户在授予同意后会被发送到该 URI。

为外部工具使用自定义标头

如果您需要在不使用身份验证管理器的情况下进行身份验证或传递请求标头，请直接在 ADK 请求中提供自定义 HTTP 标头。

在初始化 AgentRegistry 客户端时，通过提供 header_provider 回调来配置自定义标头。

header_provider 回调会接收当前执行上下文，并返回一个 HTTP 标头字典。McpToolset 组件会自动将这些标头附加到发送到 MCP 服务器的下游请求。

header_provider 回调提供的自定义标头只会发送到目标 MCP 服务器。自定义标头不用于针对 Agent Registry API 本身进行身份验证。该 API 始终依赖于 ADC。

以下示例演示了如何为外部工具集提供 API 密钥或自定义不记名令牌：

import os
from typing import Any
from google.adk.integrations.agent_registry import AgentRegistry

project_id = os.environ.get("GOOGLE_CLOUD_PROJECT")
location = os.environ.get("GOOGLE_CLOUD_LOCATION", "global")
external_api_key = os.environ.get("API_KEY")

# Define the custom header provider.
def my_auth_headers(context: Any) -> dict[str, str]:
    """Returns custom headers injected into MCP toolset requests."""
    return {
        "Authorization": f"Bearer {external_api_key}",
        "X-Custom-Context": "orchestrator-request"
    }

# Initialize the registry client.
# The header_provider is sent to the target MCP server during invocation.
# The header_provider is not used to authenticate against the Agent Registry API.
# The Agent Registry API always uses your Google ADC.
registry = AgentRegistry(
    project_id=project_id,
    location=location,
    header_provider=my_auth_headers
)

# Fetching the toolset automatically attaches the header_provider to the MCP server.
# Replace with the full resource name of your registered MCP server.
mcp_server_name = f"projects/PROJECT_ID/locations/LOCATION/mcpServers/EXTERNAL_SERVER_ID"
external_mcp_toolset = registry.get_mcp_toolset(mcp_server_name=mcp_server_name)

替换以下内容：

• PROJECT_ID：治理项目的项目 ID。
• LOCATION：注册表区域或位置，例如 us-central1。
• EXTERNAL_SERVER_ID：已注册的外部 MCP 服务器的 ID。

后续步骤

• 了解如何使用 Agent Gateway 对经过身份验证的工具请求强制执行访问权限控制和网络边界。