使用三足式 OAuth 和驗證管理員進行驗證

如要讓代理程式代表特定使用者存取外部工具和服務 (例如 Jira 工作或 GitHub 存放區),您必須在代理程式身分驗證管理員中設定三足式 OAuth 驗證提供者。

三足式 OAuth 授權供應商會為您管理使用者重新導向和權杖。這樣一來,您就不必編寫自訂程式碼來處理複雜的 OAuth 2.0 流程。

三足式 OAuth 工作流程

三足式 OAuth 驗證提供者需要使用者同意,因為代理程式會代表使用者存取資源。

  1. 提示和重新導向:聊天介面會提示使用者登入,然後將使用者重新導向至第三方應用程式的同意頁面。
  2. 同意聲明和儲存:使用者授予權限後,系統會將產生的 OAuth 權杖儲存在 Google 管理的憑證保管箱中。
  3. 注入:使用 Agent Development Kit (ADK) 時,代理會自動從驗證供應商擷取權杖,並將權杖注入工具叫用標頭。

事前準備

  1. [確認您已選擇正確的驗證方法](/iam/docs/agent-identity-overview#auth-models)。
  2. 啟用 Agent Identity Connector API。

    啟用 API 時所需的角色

    如要啟用 API,您需要服務使用情形管理員 IAM 角色 (roles/serviceusage.serviceUsageAdmin),其中包含 serviceusage.services.enable 權限。瞭解如何授予角色

    啟用 API

  3. 建立及部署代理程式
  4. 請確認您有前端應用程式,可處理使用者登入提示,並將使用者重新導向至第三方同意聲明頁面。
  5. 確認您具備完成這項工作所需的角色

必要的角色

如要取得建立及使用三方驗證供應商所需的權限,請要求系統管理員授予您專案的下列 IAM 角色:

如要進一步瞭解如何授予角色,請參閱「管理專案、資料夾和組織的存取權」。

這些預先定義的角色具備建立及使用三足式驗證供應商所需的權限。如要查看確切的必要權限,請展開「Required permissions」(必要權限) 部分:

所需權限

如要建立及使用三足式驗證供應商,必須具備下列權限:

  • 如要建立驗證提供者: iamconnectors.connectors.create
  • 如何使用驗證提供者:
    • iamconnectors.connectors.retrieveCredentials
    • aiplatform.endpoints.predict
    • aiplatform.sessions.create

您或許還可透過自訂角色或其他預先定義的角色取得這些權限。

建立 3 向驗證供應程式

建立驗證提供者,定義第三方應用程式的設定和憑證。

如要建立三方授權供應商,請使用 Google Cloud 控制台或 Google Cloud CLI。

控制台

  1. 前往 Google Cloud 控制台的「Agent Registry」頁面。

    前往 Agent Registry

  2. 按一下要為其建立驗證供應商的代理程式名稱。
  3. 按一下「身分」
  4. 在「Auth Providers」(驗證提供者) 專區中,按一下「新增驗證提供者」
  5. 在「新增驗證供應商」窗格中,輸入名稱和說明。

    名稱只能使用小寫英文字母、數字或連字號,開頭須為小寫英文字母,結尾不得為連字號。

  6. 在「OAuth Type」(OAuth 類型) 清單中,選取「OAuth (3 legged)」(OAuth (三足式))。 。
  7. 點按「Create and continue」(建立並繼續)
  8. 如要授予代理程式身分權限,允許使用驗證提供者,請按一下「Grant access」(授予存取權)。

    這會自動將「連結器使用者」(roles/iamconnectors.user) 角色指派給授權提供者資源上的代理程式身分。

  9. 複製回呼網址。
  10. 在另一個分頁中,於第三方應用程式註冊回呼網址。
  11. 從第三方應用程式取得用戶端 ID、用戶端密鑰、權杖網址和授權網址。
  12. 在「驗證供應商憑證」部分,輸入下列資訊:
    • 用戶端 ID
    • 用戶端密碼
    • 權杖網址
    • 授權網址
  13. 按一下「新增供應商設定」

新建立的驗證供應商會顯示在「驗證供應商」清單中。

Google Cloud CLI

  1. 使用授權和權杖網址建立驗證提供者:

    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"
  2. 從驗證提供者詳細資料擷取重新導向 URI:

    gcloud alpha agent-identity connectors describe AUTH_PROVIDER_NAME \
        --location="LOCATION"

    指令會在 redirectUrl 欄位中傳回重新導向 URI。

  3. 向第三方應用程式註冊重新導向 URI,取得用戶端 ID 和用戶端密鑰。

  4. 使用用戶端 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"
  5. 如要授予代理程式身分使用驗證供應商的權限,請更新專案或特定驗證供應商的 IAM 允許政策,並將連接器使用者 (roles/iamconnectors.user) 角色授予代理程式主體。

    代理身分採用業界標準的 SPIFFE ID 格式。在 IAM 允許政策中,代理程式身分是透過主體 ID參照。

    專案層級 (gcloud)

    在專案層級授予角色,可讓代理程式使用該專案中的任何驗證提供者。

    • 如要授予單一代理程式專案中驗證供應商的存取權,請執行下列指令:

      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"

    連接器層級 (捲曲)

    如要授予單一代理程式特定驗證供應商的存取權,請使用 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)。您的代理程式必須處理作業的生命週期:

  1. 初始要求:代理程式會呼叫 retrieveCredentials
  2. 需要同意聲明:如果使用者尚未授予同意聲明,API 會傳回 LRO,其中中繼資料包含 auth_uriconsent_nonce
  3. 前端重新導向:應用程式必須將使用者重新導向至 auth_uri
  4. 完成:使用者授予同意聲明後,請使用 consent_nonce 呼叫 FinalizeCredential,完成流程並取得權杖。

更新用戶端應用程式

如要處理 3-legged OAuth 的使用者登入和重新導向,用戶端應用程式必須實作下列步驟,管理使用者同意聲明並繼續對話:

如需完整實作範例,請參閱「ValidateUserId 前端範例」。

處理授權觸發程序

如果代理程式需要使用者同意,就會傳回 adk_request_credential 函式呼叫。應用程式必須攔截這項呼叫,才能啟動使用者授權對話方塊或重新導向。

記錄授權供應商提供的 consent_nonce,管理工作階段內容。驗證步驟需要這個隨機值,才能驗證使用者。在工作階段中儲存 auth_configauth_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)。這個端點必須符合下列條件:

  1. 以查詢參數形式接收 user_id_validation_stateauth_provider_name
  2. 從工作階段內容擷取 user_idconsent_nonce
  3. 使用這些參數呼叫驗證供應商的 FinalizeCredentials API。
  4. 收到成功回應後,請關閉授權視窗。
@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_configauth_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時,請務必啟用代理身分。

如要部署至 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]"],
    },
)

後續步驟