Google uses AI technology to translate content into your preferred language. AI translations can contain errors.

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

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

本文档介绍如何处理 Agent Registry 资源的身份验证。

准备工作

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

在项目中设置代理注册表。
安装或升级到最新版本的智能体开发套件 (ADK)，并包含必要的 A2A 依赖项：
pip
pip install --upgrade "google-adk[a2a]"
uv
uv add "google-adk[a2a]"
您必须升级到至少 google-adk>=1.29.0。
如需使用应用默认凭证 (ADC) 进行身份验证，请配置这些凭证：
```
gcloud auth application-default login
```
ADC 凭据必须具有代理或工具与之交互的底层 Google Cloud 服务所需的 Identity and Access Management (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 密钥。

了解身份验证模型

为了管理凭据，Agent Registry 依赖于 Agent Identity 进行内置 Google Cloud 身份验证，并依赖于 Agent Identity Auth Manager 来实现身份验证提供程序和绑定。

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

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

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

为 Google Cloud 工具使用代理身份

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

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

对远程 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 服务器时，您可以使用 Agent Identity Auth Manager 来管理凭据，而无需在应用中对其进行硬编码。

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

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

这些模型要求您创建身份验证提供程序。然后，您可以在代理注册表中创建绑定，以将身份验证提供方与注册表资源相关联。

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

访问自定义工具

如果您希望代理使用自己的凭据连接到外部工具或自定义工具，可以配置 Agent Identity Auth Manager 来处理 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 回调提供的自定义标头仅发送到目标 Google Cloud 服务（即代理或工具）。自定义标头不用于针对 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)

后续步骤

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