認証マネージャーで 3-legged OAuth を使用して認証する

特定のエンドユーザーに代わって、エージェントが外部ツールやサービス(Jira タスクや GitHub リポジトリなど)にアクセスできるようにするには、Agent Identity 認証マネージャーで 3-legged OAuth 認証プロバイダを構成します。

3-legged OAuth 認証プロバイダは、認証情報とトークンを管理することで、認証フローを処理するためのカスタムコードを不要にします。

3-legged OAuth ワークフロー

3-legged OAuth 認証プロバイダでは、エージェントがユーザーに代わってリソースにアクセスするため、ユーザーの同意が必要です。

  1. プロンプトとリダイレクト: チャット インターフェースでユーザーにログインを求め、 サードパーティ アプリケーションの同意ページにリダイレクトします。
  2. 同意と保存: ユーザーが権限を付与すると、Agent Identity 認証マネージャーは、結果の OAuth トークンを Google 管理の認証情報ボールトに保存します。
  3. 挿入: Agent Development Kit(ADK)を使用すると、エージェントは 認証プロバイダからトークンを自動的に取得し、 ツール呼び出しヘッダーに挿入します。

始める前に

  1. 正しい認証方法を選択していることを確認します
  2. Agent Identity API を有効にします。

    API を有効にするために必要なロール

    API を有効にするには、serviceusage.services.enable 権限を含む Service Usage 管理者 IAM ロール(roles/serviceusage.serviceUsageAdmin)が必要です。詳しくは、ロールを付与する方法をご覧ください。

    API の有効化

  3. エージェント を作成してデプロイします
  4. ユーザーのログイン プロンプト とサードパーティの同意ページへのリダイレクトを処理するフロントエンド アプリケーションがあることを確認します。
  5. このタスクを完了するために必要なロールがあることを確認します

必要なロール

3-legged 認証プロバイダを作成して使用するために必要な権限を取得するには、プロジェクトに対する次の IAM ロールの付与を管理者に依頼してください。

ロールの付与については、プロジェクト、フォルダ、組織へのアクセス権の管理をご覧ください。

これらの事前定義ロールには 3-legged 認証プロバイダの作成と使用に必要な権限が含まれています。必要とされる正確な権限については、「必要な権限」セクションを開いてご確認ください。

必要な権限

3-legged 認証プロバイダを作成して使用するには、次の権限が必要です。

  • 認証プロバイダを作成するには: agentidentity.authProviders.create
  • 認証プロバイダを使用するには:
    • agentidentity.authProviders.retrieveCredentials
    • aiplatform.endpoints.predict
    • aiplatform.sessions.create

カスタムロールや他の事前定義ロールを使用して、これらの権限を取得することもできます。

3-legged 認証プロバイダを作成する

認証プロバイダを作成して、サードパーティ アプリケーションの構成と認証情報を定義します。

3-legged 認証プロバイダを作成するには、gcloud CLI を使用します。

  1. OAuth クライアント アプリケーションを構成して、クライアントを登録し、クライアント ID と クライアント シークレットを取得します。その セクションのテンプレートを使用して、リダイレクト URI を指定します。

  2. クライアント認証情報を使用して認証プロバイダを作成します。

    gcloud alpha agent-identity authProviders create AUTH_PROVIDER_NAME \
        --project="PROJECT_ID" \
        --location="LOCATION" \
        --three-legged-oauth-client-id="CLIENT_ID" \
        --three-legged-oauth-client-secret="CLIENT_SECRET" \
        --three-legged-oauth-authorization-url="AUTHORIZATION_URL" \
        --three-legged-oauth-token-url="TOKEN_URL"
  3. 認証プロバイダがリストに表示され、状態が ENABLED であることを確認します。
    gcloud alpha agent-identity authProviders list \
       --project="PROJECT_ID" \
       --location="LOCATION"
  4. エージェントとローカル開発 環境が認証プロバイダから認証情報を取得できるように、アクセス権限を付与します。デプロイされたエージェントと個人のユーザー アカウントが認証プロバイダにアクセスできるようにするには、認証プロバイダ リソースにAgent Identity ユーザーroles/agentidentity.user)ロールを付与します。

    1. デプロイされたエージェントの SPIFFE ID(Agent Identity)へのアクセス権を付与します。

      gcloud alpha agent-identity authProviders add-iam-policy-binding AUTH_PROVIDER_NAME \
          --project="PROJECT_ID" \
          --location="LOCATION" \
          --role="roles/agentidentity.user" \
          --member="principal://agents.global.org-ORGANIZATION_ID.system.id.goog/resources/aiplatform/projects/PROJECT_NUMBER/locations/LOCATION/reasoningEngines/ENGINE_ID"
    2. ローカル開発とテスト (adk web)用の個人ユーザー アカウントへのアクセス権を付与します。

      gcloud alpha agent-identity authProviders add-iam-policy-binding AUTH_PROVIDER_NAME \
          --project="PROJECT_ID" \
          --location="LOCATION" \
          --role="roles/agentidentity.user" \
          --member="user:USER_EMAIL"

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

  • PROJECT_ID: 実際の Google Cloud プロジェクト ID。
  • LOCATION: 認証プロバイダとエージェントがデプロイされているロケーション(us-west1 など)。
  • AUTH_PROVIDER_NAME: 認証プロバイダの名前(bigquery-mcp-3lo-authprovider など)。
  • AUTHORIZATION_URL: 認証サーバーの URL(https://accounts.google.com/o/oauth2/v2/auth など)。
  • TOKEN_URL: トークン サーバーの URL(https://oauth2.googleapis.com/token など)。
  • CLIENT_ID: サードパーティ サービスから生成した OAuth クライアント ID。
  • CLIENT_SECRET: サードパーティ サービスから生成した OAuth クライアント シークレット。
  • ORGANIZATION_ID: 実際の Google Cloud 組織 ID。
  • PROJECT_NUMBER: 実際の Google Cloud プロジェクトの番号。
  • ENGINE_ID: デプロイされた推論エンジン エージェントの ID。
  • USER_EMAIL: 個人のユーザー アカウントのメールアドレス。

OAuth クライアント アプリケーションを構成する

OAuth クライアント認証情報を登録する前に、サードパーティの認証サーバー(Google、GitHub、Jira など)からクライアント ID とクライアント シークレットを取得します。

Google Cloud以外のサードパーティ サービスに接続する場合は、その サービスのデベロッパー ポータルから OAuth クライアント認証情報を取得し、このセクションの手順をスキップします。

リダイレクト URI を登録する

OAuth クライアント認証情報を構成するときは、認証プロバイダ専用のコールバック リダイレクト URI を登録する必要があります。

  1. 次のテンプレートを使用してリダイレクト URI を作成します。

    https://agentidentitycredentials.googleapis.com/v1alpha/projects/PROJECT_ID/locations/LOCATION/authProviders/AUTH_PROVIDER_NAME/oauthcallback

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

    • PROJECT_ID: 実際の Google Cloud プロジェクト ID。
    • LOCATION: 認証プロバイダがデプロイされるリージョン(us-west1 など)。
    • AUTH_PROVIDER_NAME: 認証プロバイダの名前。

    例: https://agentidentitycredentials.googleapis.com/v1alpha/projects/my-project/locations/us-west1/authProviders/bigquery-mcp-3lo-authprovider/oauthcallback

  2. サービス( BigQuery など)に接続する場合は、 Google Cloud コンソールで同意画面を構成し、OAuth クライアント認証情報を作成できます。 Google Cloud

    1. OAuth 同意画面を構成します。

      1. コンソールで、[API とサービス] >[OAuth 同意画面] ページに移動します。 Google Cloud

        [API とサービス] > [OAuth 同意画面] に移動

      2. [**アプリ情報**] セクションに、アプリケーション名(BigQuery Manager Application など)とサポート メールを入力します。
      3. [対象] セクションで、[内部] または [外部] を選択します。
      4. 通知を受け取るための連絡先情報を入力します。
      5. Google API サービス ユーザーデータ に関するポリシー を読んで同意します。
      6. [完了] をクリックします。
    2. OAuth クライアント認証情報を作成します。

      1. Google Cloud コンソールで、[API とサービス]>[OAuth 同意画面] >[クライアント] ページに移動します。

        [API とサービス] >[OAuth 同意画面] >[クライアント] に移動

      2. [認証情報を作成]>[OAuth クライアント ID] をクリックします。
      3. リストから [ウェブ アプリケーション] オプションを選択します。
      4. OAuth クライアントにわかりやすい名前を入力します。
      5. [**承認済みのリダイレクト URI**] セクションで、 [**URI を追加**] をクリックして、作成したリダイレクト URI を入力します。
      6. [作成] をクリックします。[OAuth クライアントが作成されました] ダイアログで、生成された [クライアント ID] と [クライアント シークレット] の値をコピーします。

エージェント コードで認証する

エージェントを認証するには、ADK を使用するか、Agent Identity API を直接呼び出します。

ADK

ADK の MCP ツールセットを使用して、エージェントのコードで認証プロバイダを参照します。

from google.adk.agents.llm_agent import LlmAgent
from google.adk.auth.credential_manager import CredentialManager
from google.adk.integrations.agent_identity import GcpAuthProvider, GcpAuthProviderScheme
from google.adk.tools.mcp_tool.mcp_session_manager import StreamableHTTPConnectionParams
from google.adk.tools.mcp_tool.mcp_toolset import McpToolset
from google.adk.auth.auth_tool import AuthConfig

# Register Google Cloud auth provider
CredentialManager.register_auth_provider(GcpAuthProvider())

# The URI to redirect the user to after consent is granted.
CONTINUE_URI = "https://YOUR_FRONTEND_URL/validateUserId"

# Create auth provider scheme
# Note: If using the legacy V1 API, the resource name uses 'connectors'
# instead of 'authProviders': projects/.../connectors/...
auth_scheme = GcpAuthProviderScheme(
    name="projects/PROJECT_ID/locations/LOCATION/authProviders/AUTH_PROVIDER_NAME",
    continue_uri=CONTINUE_URI
)

# Configure an MCP tool with the authentication scheme.
toolset = McpToolset(
    connection_params=StreamableHTTPConnectionParams(url="https://YOUR_MCP_SERVER_URL"),
    auth_scheme=auth_scheme,
)

# Initialize the agent with the authenticated tools.
agent = LlmAgent(
    name="AGENT_NAME",
    model="gemini-2.5-flash",
    instruction="AGENT_INSTRUCTIONS",
    tools=[toolset],
)

例: BigQuery MCP に接続する

次の例は、エージェントを BigQuery MCP サーバーに接続する agent.py 構成を示しています。

import os
from google.adk.agents import Agent
from google.adk.apps import App
from google.adk.auth.credential_manager import CredentialManager
from google.adk.integrations.agent_identity import GcpAuthProvider, GcpAuthProviderScheme
from google.adk.models import Gemini
from google.adk.tools.mcp_tool.mcp_session_manager import StreamableHTTPConnectionParams
from google.adk.tools.mcp_tool.mcp_toolset import McpToolset
import google.auth
from google.genai import types

_, project_id = google.auth.default()
os.environ["GOOGLE_CLOUD_PROJECT"] = "PROJECT_ID"
os.environ["GOOGLE_GENAI_USE_VERTEXAI"] = "True"

bigquery_mcp_auth_provider_id = "AUTH_PROVIDER_NAME"
bigquery_mcp_endpoint = os.environ.get(
    "BIGQUERY_MCP_ENDPOINT", "https://bigquery.googleapis.com/mcp"
)

# Register Google Cloud auth provider
CredentialManager.register_auth_provider(GcpAuthProvider())

# URI to redirect user to after consent is granted.
CONTINUE_URI = "http://127.0.0.1:8501/validateUserId"

bigquery_mcp_auth_scheme = GcpAuthProviderScheme(
    name=f"projects/{project_id}/locations/LOCATION/authProviders/{bigquery_mcp_auth_provider_id}",
    scopes=["https://www.googleapis.com/auth/bigquery"],
    continue_uri=CONTINUE_URI,
)

bigquery_mcp_tools = McpToolset(
    connection_params=StreamableHTTPConnectionParams(url=bigquery_mcp_endpoint),
    auth_scheme=bigquery_mcp_auth_scheme,
    errlog=None,
)

root_agent = Agent(
    name="root_agent",
    model=Gemini(
        model="gemini-2.5-flash",
        retry_options=types.HttpRetryOptions(attempts=3),
    ),
    instruction=(
        "You are a helpful AI assistant designed to provide accurate and useful"
        " information. You can also use your BigQuery MCP tools to look up"
        " BigQuery data."
    ),
    tools=[bigquery_mcp_tools],
)

app = App(
    root_agent=root_agent,
    name="AGENT_NAME",
)

ADK

ADK の認証済み関数ツールを使用して、エージェントのコードで認証プロバイダを参照します。

import httpx
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_identity import GcpAuthProviderScheme
from google.adk.apps import App
from google.adk.auth.auth_credential import AuthCredential
from google.adk.auth.auth_tool import AuthConfig
from google.adk.tools.authenticated_function_tool import AuthenticatedFunctionTool
from vertexai import agent_engines

# First, register Google Cloud auth provider
CredentialManager.register_auth_provider(GcpAuthProvider())

# The URI to redirect the user to after consent is completed.
CONTINUE_URI = "WEB_APP_VALIDATE_USER_URI"

# Create Auth Config
spotify_auth_config = AuthConfig(
    auth_scheme=GcpAuthProviderScheme(
        name="projects/PROJECT_ID/locations/LOCATION/authProviders/AUTH_PROVIDER_NAME",
        continue_uri=CONTINUE_URI
    )
)

# Use the Auth Config in Authenticated Function Tool
spotify_search_track_tool = AuthenticatedFunctionTool(
    func=spotify_search_track, auth_config=spotify_auth_config
)

# Sample function tool
async def spotify_search_track(credential: AuthCredential, query: str) -> str | list:
    token = None
    if credential.http and credential.http.credentials:
        token = credential.http.credentials.token

    if not token:
        return "Error: No authentication token available."

    async with httpx.AsyncClient() as client:
        response = await client.get(
            "https://api.spotify.com/v1/search",
            headers={"Authorization": f"Bearer {token}"},
            params={"q": query, "type": "track", "limit": 1},
        )
        # Add your own logic here

agent = LlmAgent(
    name="AGENT_NAME",
    model="gemini-2.5-flash",
    instruction="AGENT_INSTRUCTIONS",
    tools=[spotify_search_track_tool],
)

app = App(
    name="APP_NAME",
    root_agent=agent,
)

vertex_app = agent_engines.AdkApp(app_name=app)

ADK

ADK の Agent Registry MCP ツールセットを使用して、エージェントのコードで認証プロバイダを参照します。

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_identity import GcpAuthProviderScheme
from google.adk.tools.mcp_tool.mcp_session_manager import StreamableHTTPConnectionParams
from google.adk.tools.mcp_tool.mcp_toolset import McpToolset
from google.adk.auth.auth_tool import AuthConfig
from google.adk.integrations.agent_registry import AgentRegistry

# First, register Google Cloud auth provider
CredentialManager.register_auth_provider(GcpAuthProvider())

# The URI to redirect the user to after consent is completed.
CONTINUE_URI="WEB_APP_VALIDATE_USER_URI"

# Create Google Cloud auth provider scheme
auth_scheme = GcpAuthProviderScheme(
    name="projects/PROJECT_ID/locations/LOCATION/authProviders/AUTH_PROVIDER_NAME",
    continue_uri=CONTINUE_URI
)

# Set Agent Registry
registry = AgentRegistry(project_id="PROJECT_ID", location="global")

toolset = registry.get_mcp_toolset(
    mcp_server_name=(
        "projects/PROJECT_ID/locations/"
        "global/mcpServers/"
        "agentregistry-00000000-0000-0000-0000-000000000000"
    ),
    auth_scheme=auth_scheme,
)

# Example MCP tool
toolset = McpToolset(
    connection_params=StreamableHTTPConnectionParams(url="MCP_URL"),
    auth_scheme=auth_scheme,
)

agent = LlmAgent(
    name="AGENT_NAME",
    model="MODEL_NAME",
    instruction="AGENT_INSTRUCTIONS",
    tools=[toolset],
)

API を直接使用する

ADK を使用していない場合、エージェントは agentidentitycredentials.retrieveCredentials API を呼び出してトークンを取得する必要があります。

これは複数ステップの OAuth フローであるため、エージェントはオペレーションのライフサイクルを処理する必要があります。

  1. 最初のリクエスト: エージェントが retrieveCredentials を呼び出します。
  2. 同意が必要: ユーザーが同意していない場合、API は 結果を uri_consent_required として含むレスポンスを返します。これには authorization_uriconsent_nonce が含まれます。
  3. フロントエンドのリダイレクト: アプリケーションはユーザーを authorization_uriにリダイレクトする必要があります。
  4. 完了: ユーザーが同意したら、 FinalizeCredentials を使用して consent_nonce を 呼び出し、フローを完了してトークンを取得します。

クライアントサイド アプリケーションを更新する

3-legged OAuth のユーザーのログインとリダイレクトを処理するには、クライアントサイド アプリケーションで次の手順を実装して、ユーザーの同意を管理し、会話を再開する必要があります。

サンプル UI サーバー

uvicorn を使用する完全なサンプル UI サーバーをダウンロードして実行できます。始める前に、GitHub アカウントと pip がインストールされていることを確認してください。

サンプル UI サーバーを設定して実行する手順は次のとおりです。

  1. adk-python GitHub リポジトリのクローンを作成します。

    git clone https://github.com/google/adk-python.git
  2. リポジトリに移動して、Python 仮想 環境を有効にします。

    cd adk-python
    python3 -m venv .venv
    source .venv/bin/activate
  3. サンプル UI クライアント ディレクトリに移動します。

    cd contributing/samples/integrations/gcp_auth/client
  4. クライアントの依存関係をインストールします。

    pip install -r requirements.txt
  5. サーバーを起動する前に、AGENT_PROJECT_DIR 環境変数を設定して、エージェント コードが配置されているディレクトリを 指定します。指定しない場合、アプリケーションはデフォルトでクライアント ディレクトリの親フォルダ内のエージェントを検索します。

    uvicorn を使用してサンプル UI サーバーを起動します。ポートが OAuth クライアントで構成されたリダイレクト URI と一致していることを確認します。

    export AGENT_PROJECT_DIR="/path/to/your/agent_project"
    uvicorn main:app --port 8501 --reload
  6. ブラウザで http://localhost:8501 を開きます。(注: OAuth リダイレクト URL では localhost ホスト名が明示的に必要であるため、127.0.0.1 ではなく localhost を使用する必要があります)。設定を指定し、 [**設定を保存して適用**] をクリックして、エージェントを操作します。

カスタム UI アプリケーション

カスタム UI アプリケーションでこれらの機能を直接実装するには、次の手順を行います。

認可トリガーを処理する

エージェントがユーザーの同意を必要とする場合、adk_request_credential 関数呼び出しを返します。アプリケーションはこの呼び出しをインターセプトして、ユーザー認可ダイアログまたはリダイレクトを開始する必要があります。

認証プロバイダから提供された consent_nonce を記録して、セッション コンテキストを管理します。この nonce は、検証ステップでユーザーを確認するために必要です。ユーザーが同意した後にフローを再開できるように、セッション内に auth_config 値と auth_request_function_call_id 値を保存します。

if (fc := get_auth_request_function_call(event_data)):
    print("--> Authentication required by agent.")
    try:
        auth_config = get_auth_config(fc)
        auth_uri, consent_nonce = handle_adk_request_credential(
            auth_config, AUTH_PROVIDER_NAME, request.user_id
        )
        if auth_uri:
            event_data['popup_auth_uri'] = auth_uri
            fc_id = (
                fc.get('id') if isinstance(fc, dict)
                else getattr(fc, 'id', None)
            )
            event_data['auth_request_function_call_id'] = fc_id
            event_data['auth_config'] = auth_config.model_dump()

            # Store session state
            if session_id:
                consent_sessions[session_id] = {
                    "user_id": request.user_id,
                    "consent_nonce": consent_nonce
                }
    except Exception as e:
        print(f"Error handling adk_request_credential: {e}")
        # Optionally, add logic to inform the user about the error.

def handle_adk_request_credential(auth_config, auth_provider_name, user_id):
    ec = auth_config.exchanged_auth_credential
    if ec and ec.oauth2:
        oauth2 = ec.oauth2
        return oauth2.auth_uri, oauth2.nonce
    return None, None

ユーザー検証エンドポイントを実装する

ウェブサーバーに検証エンドポイントを実装します(構成時に continue_uri として指定した URI と同じ)。このエンドポイントは次の処理を行う必要があります。

  1. user_id_validation_stateauth_provider_nameuuid をクエリ パラメータとして受け取ります。
  2. セッション ストレージから user_id 値と consent_nonce 値 を取得します。複数の認可フローが同時に実行される場合は、 uuid を使用して正しいセッションを照合します。
  3. これらの パラメータを使用して、認証プロバイダの FinalizeCredentials API を呼び出します。
  4. 成功レスポンスを受け取ったら、認可ウィンドウを閉じます。
例: FastAPI 検証エンドポイント(main.py

次の例は、OAuth コールバックを処理してユーザー認証情報を確定する完全な FastAPI 検証エンドポイントを示しています。

@app.api_route("/validateUserId", methods=["GET"])
async def validate_user(request: Request):
    auth_provider_name = request.query_params.get("auth_provider_name")
    session_id = request.cookies.get("session_id")
    session = consent_sessions.get(session_id, {})

    payload = {
        "userId": session.get("user_id"),
        "userIdValidationState": request.query_params.get(
            "user_id_validation_state"
        ),
        "consentNonce": session.get("consent_nonce"),
    }

    base_url = "https://agentidentitycredentials.googleapis.com/v1alpha"
    finalize_url = f"{base_url}/{auth_provider_name}/credentials:finalize"

    try:
        async with httpx.AsyncClient(timeout=30.0) as client:
            resp = await client.post(finalize_url, json=payload)
            resp.raise_for_status()
    except httpx.HTTPError as e:
        err_text = e.response.text if hasattr(e, "response") else str(e)
        status = e.response.status_code if hasattr(e, "response") else 500
        return HTMLResponse(err_text, status_code=status)

    return HTMLResponse("""
        <script>
            window.close();
        </script>
        <p>Success. You can close this window.</p>
    """)

エージェントとの会話を再開する

ユーザーが同意して認可ウィンドウが閉じたら、セッション データから auth_config 値と auth_request_function_call_id 値を取得します。会話を続けるには、これらの詳細を function_response としてエージェントへの新しいリクエストに含めます。

if (request.is_auth_resume and session.auth_request_function_call_id
    and session.auth_config):
    auth_content = types.Content(
        role='user',
        parts=[
            types.Part(
                function_response=types.FunctionResponse(
                    id=session.auth_request_function_call_id,
                    name='adk_request_credential',
                    response=session.auth_config
                )
            )
        ],
    )
    # Send message to agent
    async for event in agent.async_stream_query(
        user_id=request.user_id,
        message=auth_content,
        session_id=session_id,
    ):
        # ...

エージェントをデプロイする

にエージェントをデプロイするときは、Agent Identity が 有効になっていることを確認してください。 Google Cloud

Gemini Enterprise Agent Platform の Agent Runtime にデプロイする場合は、 identity_type=AGENT_IDENTITY フラグを使用します。

import vertexai
from vertexai import types
from vertexai.agent_engines import AdkApp

# Initialize the Vertex AI client with v1beta1 API for Agent Identity support
client = vertexai.Client(
    project="PROJECT_ID",
    location="LOCATION",
    http_options=dict(api_version="v1beta1")
)

# Use the proper wrapper class for your Agent Framework (e.g., AdkApp)
app = AdkApp(agent=agent)

# Deploy the agent with Agent Identity enabled
remote_app = client.agent_engines.create(
    agent=app,
    config={
        "identity_type": types.IdentityType.AGENT_IDENTITY,
        "requirements": [
            "google-cloud-aiplatform[agent_engines,adk]",
            "google-adk[agent-identity]"
        ],
    },
)

次のステップ