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

オーケストレーターエージェントが、エージェントレジストリで検出されたリモートエージェントまたは 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 キー。

認証モデルについて学習する

認証情報を管理するために、エージェントレジストリは、組み込みの Google Cloud 認証にエージェント ID を使用し、認証プロバイダとバインディングにエージェント ID 認証マネージャーを使用します。

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

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

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

Google Cloud ツールにエージェント ID を使用する

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

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

リモート 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 サーバーに接続する必要がある場合は、エージェント ID 認証マネージャーを使用して、アプリケーションに認証情報をハードコードせずに管理できます。

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

カスタムツールにアクセスする: エージェントが独自の ID と権限を使用してツールにアクセスする必要がある場合は、API キーまたは 2 レッグ OAuth（2LO）を使用してエージェントを認証します。
ユーザーに代わってツールにアクセスする: 3-legged OAuth（3LO）を使用して、個々のユーザーとして機能するエージェントを認証します。このモデルは、エージェントがユーザーの代わりにツールにアクセスする必要がある場合に推奨されます。この場合、ユーザーの同意と委任された権限が必要になります。

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

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

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

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

認証プロバイダを作成する: エージェント ID 認証マネージャーで認証プロバイダを作成して、API キーまたは OAuth 認証情報を保存します。詳細については、認証マネージャーで 2-legged OAuth を使用して認証するまたは認証マネージャーで API キーを使用して認証するをご覧ください。
認証プロバイダをバインドする: オーケストレーターエージェントがターゲットの正しい認証プロバイダを検出できるようにするには、エージェントと認証プロバイダの間の 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 サーバーまたはツールに対して認証を行う場合は、エージェント ID 認証マネージャーを介して 3-legged OAuth（3LO）を使用します。

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

認証プロバイダを作成する: Agent Identity 認証マネージャーで認証プロバイダを作成し、OAuth リダイレクト URI を構成します。詳細については、認証マネージャーで 3-legged OAuth を使用して認証するをご覧ください。
クライアントアプリケーションを更新する: adk_request_credential 関数呼び出しを処理し、ユーザーの同意を管理するように、アプリケーションクライアントを更新します。詳細な手順については、クライアントサイドアプリケーションを更新するをご覧ください。
認証プロバイダをバインドする: オーケストレーターエージェントがターゲットの正しい認証プロバイダを検出できるようにするには、エージェントと認証プロバイダの間の 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 コードでバインディングを解決する

エージェントレジストリで認証プロバイダバインディングを作成すると、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)

次のステップ

Agent Gateway を使用して、認証されたツールリクエストにアクセス制御とネットワーク境界を適用する方法について説明します。