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

エージェントが特定のユーザーに代わって外部ツールやサービス（Jira タスクや GitHub リポジトリなど）にアクセスする場合は、エージェント ID 認証マネージャーで 3-legged OAuth 認証プロバイダを構成する必要があります。

3-legged OAuth 認証プロバイダは、ユーザーのリダイレクトとトークンを管理します。これにより、複雑な OAuth 2.0 フローを処理するカスタムコードを作成する必要がなくなります。

3-legged OAuth ワークフロー

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

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

始める前に

[正しい認証方法を選択していることを確認します](/iam/docs/agent-identity-overview#auth-models)。
Agent Identity Connector API を有効にします。
API を有効にするために必要なロール
API を有効にするには、serviceusage.services.enable 権限を含む Service Usage 管理者 IAM ロール（roles/serviceusage.serviceUsageAdmin）が必要です。詳しくは、ロールを付与する方法をご覧ください。
API の有効化
エージェントを作成してデプロイする。
ユーザーのログインプロンプトとサードパーティの同意ページへのリダイレクトを処理するフロントエンドアプリケーションがあることを確認します。
このタスクを完了するために必要なロールがあることを確認します。

必要なロール

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

認証プロバイダを作成するには:
- IAM Connector 管理者（roles/iamconnectors.admin）
- IAM Connector 編集者（roles/iamconnectors.editor）
認証プロバイダを使用するには:
- IAM Connector ユーザー（roles/iamconnectors.user）
- エージェントのデフォルトアクセス（roles/aiplatform.agentDefaultAccess）
- エージェントコンテキスト編集者（roles/aiplatform.agentContextEditor）
- Vertex AI ユーザー（roles/aiplatform.user）
- Service Usage ユーザー（roles/serviceusage.serviceUsageConsumer）

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

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

必要な権限

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

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

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

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

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

3-legged 認証プロバイダを作成するには、 Google Cloud コンソールまたは Google Cloud CLI を使用します。

コンソール

コンソールで、[エージェントレジストリ] ページに移動します。 Google Cloud
エージェントレジストリに移動
認証プロバイダを作成するエージェントの名前をクリックします。
[ID] をクリックします。
[認証プロバイダ] セクションで、[ 認証プロバイダを追加] をクリックします。
[認証プロバイダを追加] パネルで、名前と説明を入力します。

名前には小文字、数字、ハイフンのみを使用できます。ハイフンで終わることはできません。小文字で始める必要があります。
[OAuth タイプ] リストから [OAuth（3-legged）] を選択します。
[作成して続行] をクリックします。
認証プロバイダを使用する権限をエージェント ID に付与するには、[アクセス権を付与] をクリックします。
これにより、認証プロバイダリソースのエージェント ID にコネクタユーザー （roles/iamconnectors.user）ロールが自動的に割り当てられます。
コールバック URL をコピーします。
別のタブで、サードパーティアプリケーションにコールバック URL を登録します。
サードパーティアプリケーションから、クライアント ID、クライアントシークレット、トークン URL、認可 URL を取得します。
[認証プロバイダの認証情報] セクションに、次の情報を入力します。
- クライアント ID
- クライアントシークレット
- トークン URL
- 認可 URL
[プロバイダ構成を追加] をクリックします。

新しく作成した認証プロバイダが [認証プロバイダ] リストに表示されます。

Google Cloud CLI

認可 URL とトークン URL を使用して認証プロバイダを作成します。

gcloud alpha agent-identity connectors create AUTH_PROVIDER_NAME \
    --location="LOCATION" \
    --three-legged-oauth-authorization-url="AUTHORIZATION_URL" \
    --three-legged-oauth-token-url="TOKEN_URL"

認証プロバイダの詳細からリダイレクト URI を取得します。
```
gcloud alpha agent-identity connectors describe AUTH_PROVIDER_NAME \
    --location="LOCATION"
```
このコマンドは、redirectUrl フィールドにリダイレクト URI を返します。
リダイレクト URI をサードパーティアプリケーションに登録して、クライアント ID とクライアントシークレットを取得します。

クライアント ID とクライアントシークレットで認証プロバイダを更新します。

gcloud alpha agent-identity connectors update AUTH_PROVIDER_NAME \
    --location="LOCATION" \
    --three-legged-oauth-client-id="CLIENT_ID" \
    --three-legged-oauth-client-secret="CLIENT_SECRET"

認証プロバイダを使用する権限をエージェント ID に付与するには、プロジェクトまたは特定の認証プロバイダの IAM 許可ポリシーを更新し、エージェントプリンシパルにコネクタユーザー （roles/iamconnectors.user）ロールを付与します。

エージェント ID は、業界標準の SPIFFE ID 形式に基づいています。IAM 許可ポリシーでは、エージェント ID はプリンシパル識別子 を使用して参照されます。
プロジェクトレベル（gcloud）

プロジェクトレベルでロールを付与すると、エージェントはそのプロジェクト内の任意の認証プロバイダを使用できます。
- プロジェクト内の認証プロバイダへのアクセス権を 1 つのエージェントに付与するには、次のコマンドを実行します。
  gcloud projects add-iam-policy-binding PROJECT_ID \ --role='roles/iamconnectors.user' \ --member="principal://agents.global.org-ORGANIZATION_ID.system.id.goog/resources/aiplatform/projects/PROJECT_NUMBER/locations/LOCATION/reasoningEngines/ENGINE_ID"
- プロジェクト内のすべてのエージェントに認証プロバイダへのアクセス権を付与するには、次のコマンドを実行します。
  gcloud projects add-iam-policy-binding PROJECT_ID \ --role='roles/iamconnectors.user' \ --member="principalSet://agents.global.org-ORGANIZATION_ID.system.id.goog/attribute.platformContainer/aiplatform/projects/PROJECT_NUMBER"
コネクタレベル（curl）

特定のエージェントに特定の認証プロバイダへのアクセス権を付与するには、 setIamPolicy API を使用します。このコマンドを実行すると、リソースの既存の許可ポリシーが上書きされます。
```
curl -X POST \
    -H "Authorization: Bearer $(gcloud auth print-access-token)" \
    -H "Content-Type: application/json" \
    -d '{
  "policy": {
    "bindings": [
      {
        "role": "roles/iamconnectors.user",
        "members": ["principal://agents.global.org-ORGANIZATION_ID.system.id.goog/resources/aiplatform/projects/PROJECT_NUMBER/locations/LOCATION/reasoningEngines/ENGINE_ID"]
      }
    ]
  }
}' \
    "https://iamconnectors.googleapis.com/v1alpha/projects/PROJECT_ID/locations/LOCATION/connectors/AUTH_PROVIDER_NAME:setIamPolicy"
```
次のように置き換えます。
- PROJECT_ID: 実際の Google Cloud プロジェクト ID。
- AUTH_PROVIDER_NAME: 認証プロバイダの名前。
- ORGANIZATION_ID: 実際の Google Cloud 組織 ID。
- PROJECT_NUMBER: あなたの Google Cloud プロジェクトの番号。
- LOCATION: エージェントのロケーション（us-central1 など）。
- ENGINE_ID: 推論エンジンの 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 the Google Cloud Auth Provider so the CredentialManager can use it.
CredentialManager.register_auth_provider(GcpAuthProvider())

# The URI to redirect the user to after consent is granted and the
# callback is received by the auth provider.
CONTINUE_URI = "https://YOUR_FRONTEND_URL/validateUserId"

# Create the Auth Provider scheme using the auth provider's full resource name.
auth_scheme = GcpAuthProviderScheme(
    name="projects/PROJECT_ID/locations/LOCATION/connectors/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="YOUR_AGENT_NAME",
    model="gemini-2.0-flash",
    instruction="YOUR_AGENT_INSTRUCTIONS",
    tools=[toolset],
)

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/connectors/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="YOUR_AGENT_NAME",
    model="YOUR_MODEL_NAME",
    instruction="YOUR_AGENT_INSTRUCTIONS",
    tools=[spotify_search_track_tool],
)

app = App(
    name="YOUR_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 by providing auth provider full resource name
auth_scheme=GcpAuthProviderScheme(
name="projects/GOOGLE_PROJECT/locations/LOCATION/connectors/AUTH_PROVIDER_NAME",
continue_uri=CONTINUE_URI)

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

toolset = registry.get_mcp_toolset(mcp_server_name="projects/GOOGLE_PROJECT/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="YOUR_AGENT_NAME",
    model="YOUR_MODEL_NAME",
    instruction="YOUR_AGENT_INSTRUCTIONS",
    tools=[toolset],
)

API を直接使用する

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

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

最初のリクエスト: エージェントが retrieveCredentials を呼び出します。
同意が必要: ユーザーが同意していない場合、API はメタデータに auth_uri と consent_nonce を含む LRO を返します。
フロントエンドのリダイレクト: アプリケーションはユーザーを auth_uriにリダイレクトする必要があります。
完了: ユーザーが同意したら、 FinalizeCredential を使用してフローを完了し、トークンを取得します。consent_nonce

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

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

認可トリガーを処理する
ユーザー検証エンドポイントを実装する
エージェントの会話を再開する

完全な実装例については、 ValidateUserId フロントエンドサンプルをご覧ください。

認可トリガーを処理する

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

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

if (auth_request_function_call := get_auth_request_function_call(event_data)):
    print("--> Authentication required by agent.")
    try:
        auth_config = get_auth_config(auth_request_function_call)
        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 = auth_request_function_call.get('id') if isinstance(auth_request_function_call, dict) else getattr(auth_request_function_call, '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):
    if auth_config.exchanged_auth_credential and auth_config.exchanged_auth_credential.oauth2:
        oauth2 = auth_config.exchanged_auth_credential.oauth2
        return oauth2.auth_uri, oauth2.nonce
    return None, None

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

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

user_id_validation_state と auth_provider_name をクエリパラメータとして受け取ります。
セッションコンテキストから user_id と consent_nonce を取得します。
これらのパラメータを使用して、認証プロバイダの FinalizeCredentials API を呼び出します。
成功レスポンスを受け取ったら、認可ウィンドウを閉じます。

@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"),
  }

  finalize_url = f"https://iamconnectorcredentials.googleapis.com/v1alpha/{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,
    ):
        # ...

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

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

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

認証マネージャーで 3-legged OAuth を使用して認証する コレクションでコンテンツを整理 必要に応じて、コンテンツの保存と分類を行います。

3-legged OAuth ワークフロー

始める前に

必要なロール

必要な権限

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

コンソール

Google Cloud CLI

プロジェクト レベル（gcloud）

コネクタ レベル（curl）

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

ADK

ADK

ADK

API を直接使用する

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