ツールとリソースに対する認証

オーケストレータ エージェントが Agent Registry を介して検出されたリモート エージェントまたは Model Context Protocol (MCP) ツールセットを呼び出す場合は、これらの機能を提供する 基盤となる Google Cloud サービスで認証を行う必要があります。

このドキュメントでは、Agent Registry リソースの認証を処理する方法について説明します。

始める前に

ツールとリソースに対して認証を行う前に、次の操作を完了してください。

  1. プロジェクトで Agent Registry を設定します。

  2. 必要な A2A 依存関係を使用して、ADK をインストールするか、最新バージョンにアップグレードします。

    pip 

    pip install --upgrade "google-adk[a2a]"

    uv 

    uv add "google-adk[a2a]"

    google-adk>=1.29.0 以上にアップグレードする必要があります。

  3. 認証にアプリケーションのデフォルト認証情報（ADC） を使用する場合は、ADC を構成します。

    gcloud auth application-default login

    ADC 認証情報には、エージェントまたはツールがやり取りする 基盤となる Google Cloud サービスに必要な IAM 権限が必要です。

  4. このガイドの例に沿って操作するには、環境変数を設定する必要があります。例:

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

    次のように置き換えます。

認証モデルについて

認証情報を管理するために、Agent Registry は組み込み認証に Agent Identity を使用し、認証プロバイダとバインディングに Agent Identity 認証マネージャー を使用します。 Google Cloud

エージェントは、独自の ID を使用することも、認証マネージャーを使用して API キーまたは OAuth でツールにアクセスすることもできます。エージェント独自の ID は、 アプリケーションのデフォルト認証情報（ADC） を介して提供されます。Agent Identity を使用しているか サービス アカウントを使用しているかにかかわらず、

または、ターゲット エージェントに高度なコンテキストを送信する場合や、外部 ツールセットが Agent Identity 認証マネージャーをサポートしていない場合は、ADK リクエストで カスタム HTTP ヘッ徊ーを直接指定できます。

ユースケースに最適なアプローチを判断するには、 認証モデルをご覧ください。

ツールに Agent Identity を使用する Google Cloud

Vertex AI や Cloud Storage などの Google Cloud サービスの場合、 エージェントは独自の ID を使用して ツールにアクセスします。このアクセスは、 アプリケーションのデフォルト認証情報（ADC） を介して、Agent Identityを使用しているか サービスアカウントを使用しているかにかかわらず、管理されます。

エージェントに関連付けられた ID には、基盤となるサービスに必要な Identity and Access Management 権限が必要です。 Google Cloud たとえば、MCP ツールが Compute Engine VM インスタンスを管理する場合、エージェント ID には、 Compute インスタンス管理者（v1）以上のロールに加えて、Agent Registry API 閲覧者ロールが必要です。

リモート A2A エージェントを認証する

オーケストレータ エージェントが独自の ID を使用してリモート 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 ツールを認証する

オーケストレータ エージェントが独自の ID を使用して 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 認証マネージャー を使用して、アプリケーションにハードコードせずに認証情報を管理できます。

ユースケースに応じて、次のシナリオで認証マネージャーを構成できます。

これらのモデルでは、認証プロバイダを作成する必要があります 。次に、Agent Registry で バインディングを作成して、認証 プロバイダをレジストリリソースにリンクします。

認証プロバイダを構成すると、ADK はオーケストレータの認証フローを管理します。例については、 ADK コードでバインディングを解決するをご覧ください。

カスタムツールにアクセスする

エージェントが独自の認証情報を使用して外部ツールまたはカスタムツールに接続する場合は、API キーまたは 2-legged OAuth（2LO）を処理するように Agent Identity 認証マネージャーを構成できます。

  1. 認証プロバイダを作成する: Agent Identity 認証マネージャーで認証プロバイダを作成して、API キーまたは OAuth 認証情報を保存します。詳細については、 認証マネージャーで 2-legged OAuth を使用して認証する または認証マネージャーで API キーを使用して認証するをご覧ください。

  2. 認証プロバイダをバインドする: オーケストレータ エージェントがターゲットの 正しい認証プロバイダを検出できるようにするには、エージェントと認証プロバイダの間に Agent Registry で バインディングを作成する必要があります。バインディングを使用すると、コードで認証プロバイダを手動で定義する必要がなくなります。

    バインディングを作成して、エージェントを認証プロバイダに明示的にリンクします。--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 サーバーまたはツールに対して認証を行う場合は、Agent Identity 認証マネージャーを介して 3-legged OAuth（3LO）を使用します。

認証マネージャーは、OAuth トークン、ユーザーの同意、リダイレクトを処理するためのフルマネージド Google サービスを提供します。エージェントを操作して、委任された権限を必要とするツールをトリガーすると、プラットフォームは自動的にユーザーに同意を求め、認証情報を保存してワークフローを再開します。

  1. 認証プロバイダを作成する: Agent Identity 認証マネージャーで認証プロバイダを作成し、OAuth リダイレクト URI を構成します。詳細については、 認証マネージャーで 3-legged OAuth を使用して認証するをご覧ください。
  2. クライアント アプリケーションを更新する: adk_request_credential 関数呼び出しを処理してユーザーの同意を管理するように、アプリケーション クライアントを更新します。詳細な手順については、 クライアントサイド アプリケーションを更新するをご覧ください。

  3. 認証プロバイダをバインドする: オーケストレータ エージェントがターゲットの 正しい認証プロバイダを検出できるようにするには、エージェントと認証プロバイダの間に Agent Registry で バインディングを作成する必要があります。バインディングを使用すると、コードで認証プロバイダを手動で定義する必要がなくなります。

    バインディングを作成して、エージェントを認証プロバイダに明示的にリンクします。--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 コードでバインディングを解決する

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

ユーザーに代わってツールにアクセスするために 3LO を設定する場合は、 も 指定する必要があります。continue_uriget_mcp_toolsetこの 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 コールバックによって提供されるカスタム ヘッダーは、 ターゲット 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 を使用して、認証済みツール リクエストにアクセス制御とネットワーク境界を適用する方法を学習する。