驗證工具和資源

當協調器代理程式叫用透過 Agent Registry 探索到的遠端代理程式或 Model Context Protocol (MCP) 工具集時，必須向提供這些功能的基礎服務進行驗證。

本文說明如何處理 Agent Registry 資源的驗證作業。

事前準備

驗證工具和資源之前，請先完成下列步驟：

在專案中設定 Agent Registry。
安裝或升級至最新版 Agent Development Kit (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 金鑰。

瞭解驗證模式

如要管理憑證，Agent Registry 會依賴 Agent Identity 進行內建 Google Cloud 驗證，並依賴 Agent Identity 驗證管理員進行驗證提供者和繫結。

專員可以使用自己的身分，也可以使用授權管理工具，透過 API 金鑰或 OAuth 存取工具。無論您是使用代理身分或服務帳戶，代理本身的 ID 都會透過應用程式預設憑證 (ADC) 提供。

或者，如要將進階內容傳送至目標代理程式，或外部工具集不支援代理程式身分驗證管理員，您可以在 ADK 要求中直接提供自訂 HTTP 標頭。

如要根據您的用途找出最佳方法，請參閱「驗證模型」。

使用代理身分存取 Google Cloud 工具

對於 Vertex AI 或 Cloud Storage 等 Google Cloud 服務，代理會使用自己的身分存取工具。無論您是使用代理身分或服務帳戶，這項存取權都是透過應用程式預設憑證 (ADC) 管理。

與代理程式相關聯的身分必須具備基礎服務的必要 Identity and Access Management 權限。舉例來說，如果 MCP 工具管理 Compute Engine VM 執行個體，除了代理程式登錄 API 檢視者角色，代理程式身分也必須具備Compute 執行個體管理員 (v1) 或更高版本等角色。

驗證遠端 A2A 代理程式

如果協調器代理程式使用自己的身分，連線至遠端 Google Agent2Agent (A2A) 代理程式，您必須明確加入 httpx.AsyncClient，並使用 Google 驗證標頭設定 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 金鑰或 2-legged OAuth (2LO) 驗證代理程式。
代表使用者存取工具：使用三足式 OAuth (3LO) 驗證代理程式，讓代理程式以個別使用者的身分運作。如果代理程式必須代表與其互動的人員存取工具，且需要使用者同意聲明和委派權限，建議採用這個模型。

如要使用這些模型，您必須建立驗證供應商。接著，您可以在代理程式登錄中建立繫結，將驗證供應商連結至登錄資源。

設定驗證供應商後，ADK 會管理編排器的驗證流程。如需範例，請參閱「在 ADK 程式碼中解析繫結」。

存取自訂工具

如要讓代理程式使用自己的憑證連線至外部或自訂工具，可以設定代理程式身分驗證管理員來處理 API 金鑰或雙向 OAuth (2LO)：

建立驗證供應商：在 Agent Identity 驗證管理工具中建立驗證供應商，以儲存 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 權杖、使用者同意聲明和重新導向。當您與代理程式互動並觸發需要委派權限的工具時，平台會自動提示使用者同意授權、儲存憑證，然後繼續執行工作流程：

建立驗證提供者：在 Agent Identity 驗證管理員中建立驗證提供者，並設定 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],
)

為外部工具使用自訂標頭

如要將進階脈絡傳送至目標代理程式，或外部工具組不支援 Agent Identity 驗證管理員，您可以直接在 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)

後續步驟

瞭解如何使用代理程式閘道，對已驗證的工具要求強制執行存取權控管和網路範圍。