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

当编排器代理调用通过代理注册表发现的远程代理或 Model Context Protocol (MCP) 工具集时，它必须使用提供这些功能的底层服务进行身份验证。

本文档介绍如何处理代理注册表资源的身份验证。

准备工作

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

在您的项目中设置代理注册表。
安装或升级到最新版本的智能体开发套件 (ADK)：
```
pip install --upgrade google-adk
```
您必须至少升级到 google-adk>=1.29.0。
如需使用应用默认凭证 (ADC) 进行身份验证，请对其进行配置：
```
gcloud auth application-default login
```
ADC 凭据必须具有代理或工具与之交互的底层服务所需的 IAM 权限。
如需按照本指南中的示例操作，您必须设置环境变量，例如：
```
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 密钥。

了解身份验证模型

如需管理凭据，代理注册表依赖于代理身份进行内置 Google Cloud 身份验证，并依赖于代理身份验证管理器进行身份验证提供方和绑定。

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

或者，如需向目标代理发送高级上下文，或者如果外部工具集不支持代理身份验证管理器，您可以提供自定义 HTTP 标头，直接在 ADK 请求中。

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

将代理身份用于 Google Cloud 工具

对于 Google Cloud Vertex AI 或 Cloud Storage 等服务，您的代理使用自己的身份访问工具。无论您使用的是代理身份还是服务账号，此访问权限都是通过应用默认凭证 (ADC) 进行管理的。

与您的代理关联的身份必须具有底层服务所需的 Identity and Access Management 权限。例如，如果 MCP 工具管理 Compute Engine 虚拟机实例，则代理身份必须具有 Compute Instance Admin (v1) 或更高级别的角色，以及 Agent Registry API Viewer 角色。

对远程 A2A 代理进行身份验证

如果您的编排器代理使用自己的身份连接到远程 Google Agent2Agent (A2A) 代理，您必须明确将配置了 Google 身份验证标头的 httpx.AsyncClient 添加到 get_remote_a2a_agent() 方法中。

我们建议您定义自定义超时，以防止超出默认 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,
)

对 MCP 工具进行身份验证

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

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

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

将身份验证管理器用于自定义工具和委托的访问权限

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

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

**访问自定义工具**：当代理需要使用自己的身份和权限访问工具时，使用 API 密钥或双方模式 OAuth (2LO) 对代理进行身份验证。
**代表用户访问工具** ：使用三方模式 OAuth (3LO) 对充当个人用户的代理进行身份验证。如果代理必须代表与它交互的人员访问工具，并且需要用户同意和委托权限，建议使用此模型。

这些模型要求您创建身份验证提供方。然后，您可以在代理注册表中创建一个绑定，将身份验证提供方链接到您的注册表资源。

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

访问自定义工具

如果您希望代理使用自己的凭据连接到外部工具或自定义工具，您可以配置代理身份验证管理器来处理 API 密钥或双方模式 OAuth (2LO)：

创建身份验证提供方 ：在代理身份验证管理器中创建一个身份验证提供方，以存储 API 密钥或 OAuth 凭据。如需了解相关信息，请参阅使用身份验证管理器通过双方模式 OAuth 进行身份验证或使用身份验证管理器通过 API 密钥进行身份验证。
绑定身份验证提供方：如需让编排器代理检测目标对象的正确身份验证提供方，您必须在代理注册表中在代理和身份验证提供方之间创建一个绑定。通过绑定，您无需在代码中手动定义身份验证提供方。

创建一个绑定，将您的代理明确链接到身份验证提供方。指定 --auth-provider 资源名称时，您必须使用您的项目 ID：
```
gcloud alpha agent-registry bindings create BINDING_NAME \
--project=PROJECT_ID \
--location=LOCATION \
--display-name="DISPLAY_NAME" \
--source-identifier="SOURCE_ID" \
--auth-provider="projects/PROJECT_ID/locations/LOCATION/connectors/AUTH_PROVIDER_ID" \
```
如需详细了解如何管理这些连接，请参阅管理绑定。

代表用户访问工具

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

身份验证管理器提供完全由 Google 代管式服务，用于处理 OAuth 令牌、用户同意和重定向。当您与代理交互并触发需要委托权限的工具时，平台会自动提示用户征求同意、存储凭据并恢复工作流：

创建身份验证提供方 ：在代理身份验证管理器中创建一个身份验证提供方，并配置 OAuth 重定向 URI。如需了解相关信息，请参阅使用身份验证管理器通过三方模式 OAuth 进行身份验证。
更新客户端应用 ：更新应用客户端以处理 adk_request_credential 函数调用并管理用户同意。如需了解详细说明，请参阅更新客户端应用。
绑定身份验证提供方：如需让编排器代理检测目标对象的正确身份验证提供方，您必须在代理注册表中在代理和身份验证提供方之间创建一个绑定。通过绑定，您无需在代码中手动定义身份验证提供方。

创建一个绑定，将您的代理明确链接到身份验证提供方。指定 --auth-provider 资源名称时，您必须使用您的项目 ID：
```
gcloud alpha agent-registry bindings create BINDING_NAME \
--project=PROJECT_ID \
--location=LOCATION \
--display-name="DISPLAY_NAME" \
--source-identifier="SOURCE_ID" \
--auth-provider="projects/PROJECT_ID/locations/LOCATION/connectors/AUTH_PROVIDER_ID" \
```
如需详细了解如何管理这些连接，请参阅管理绑定。

在 ADK 代码中解析绑定

在代理注册表中创建身份验证提供方绑定后，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],
)

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

[...]

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

将自定义标头用于外部工具

如需向目标代理发送高级上下文，或者如果外部工具集不支持代理身份验证管理器，您可以直接在 ADK 请求中提供自定义 HTTP 标头。

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

header_provider 回调接收当前执行上下文并返回 HTTP 标头的字典。RemoteA2aAgent 或 McpToolset 组件会自动将这些标头附加到下游请求。

由 header_provider 回调提供的自定义标头仅发送到目标服务（代理或工具）。自定义标头不用于针对 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 the tool or agent requests."""
    return {
        "Authorization": f"Bearer {external_api_key}",
        "X-Custom-Context": "orchestrator-request"
    }

# Initialize the registry client.
# The header_provider is sent to the MCP server or agent 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)

后续步骤

了解如何使用代理网关对经过身份验证的工具请求强制执行访问控制和网络边界。