如要讓代理程式代表特定使用者存取外部工具和服務 (例如 Jira 工作或 GitHub 存放區),您必須在代理程式身分驗證管理員中設定三足式 OAuth 驗證提供者。
三足式 OAuth 授權供應商會為您管理使用者重新導向和權杖。這樣一來,您就不必編寫自訂程式碼來處理複雜的 OAuth 2.0 流程。
三足式 OAuth 工作流程
三足式 OAuth 驗證提供者需要使用者同意,因為代理程式會代表使用者存取資源。
- 提示和重新導向:聊天介面會提示使用者登入,然後將使用者重新導向至第三方應用程式的同意頁面。
- 同意聲明和儲存:使用者授予權限後,系統會將產生的 OAuth 權杖儲存在 Google 管理的憑證保管箱中。
- 注入:使用 Agent Development Kit (ADK) 時,代理會自動從驗證供應商擷取權杖,並將權杖注入工具叫用標頭。
事前準備
- [確認您已選擇正確的驗證方法](/iam/docs/agent-identity-overview#auth-models)。
-
啟用 Agent Identity Connector API。
啟用 API 時所需的角色
如要啟用 API,您需要服務使用情形管理員 IAM 角色 (
roles/serviceusage.serviceUsageAdmin),其中包含serviceusage.services.enable權限。瞭解如何授予角色。 - 建立及部署代理程式。
- 請確認您有前端應用程式,可處理使用者登入提示,並將使用者重新導向至第三方同意聲明頁面。
- 確認您具備完成這項工作所需的角色。
必要的角色
如要取得建立及使用三方驗證供應商所需的權限,請要求系統管理員授予您專案的下列 IAM 角色:
-
如要建立驗證提供者:
-
IAM 連接器管理員 (
roles/iamconnectors.admin) -
IAM 連接器編輯者 (
roles/iamconnectors.editor)
-
IAM 連接器管理員 (
-
如何使用驗證提供者:
-
IAM 連接器使用者 (
roles/iamconnectors.user) -
代理人預設存取權 (
roles/aiplatform.agentDefaultAccess) -
Agent Context 編輯者 (
roles/aiplatform.agentContextEditor) -
Vertex AI 使用者 (
roles/aiplatform.user) -
服務使用情形消費者 (
roles/serviceusage.serviceUsageConsumer)
-
IAM 連接器使用者 (
如要進一步瞭解如何授予角色,請參閱「管理專案、資料夾和組織的存取權」。
這些預先定義的角色具備建立及使用三足式驗證供應商所需的權限。如要查看確切的必要權限,請展開「Required permissions」(必要權限) 部分:
所需權限
如要建立及使用三足式驗證供應商,必須具備下列權限:
-
如要建立驗證提供者:
iamconnectors.connectors.create -
如何使用驗證提供者:
-
iamconnectors.connectors.retrieveCredentials -
aiplatform.endpoints.predict -
aiplatform.sessions.create
-
建立 3 向驗證供應程式
建立驗證提供者,定義第三方應用程式的設定和憑證。
如要建立三方授權供應商,請使用 Google Cloud 控制台或 Google Cloud CLI。
控制台
- 前往 Google Cloud 控制台的「Agent Registry」頁面。
- 按一下要為其建立驗證供應商的代理程式名稱。
- 按一下「身分」。
- 在「Auth Providers」(驗證提供者) 專區中,按一下「新增驗證提供者」。
-
在「新增驗證供應商」窗格中,輸入名稱和說明。
名稱只能使用小寫英文字母、數字或連字號,開頭須為小寫英文字母,結尾不得為連字號。
- 在「OAuth Type」(OAuth 類型) 清單中,選取「OAuth (3 legged)」(OAuth (三足式))。 。
- 點按「Create and continue」(建立並繼續)。
- 如要授予代理程式身分權限,允許使用驗證提供者,請按一下「Grant access」(授予存取權)。
這會自動將「連結器使用者」(
roles/iamconnectors.user) 角色指派給授權提供者資源上的代理程式身分。 - 複製回呼網址。
- 在另一個分頁中,於第三方應用程式註冊回呼網址。
- 從第三方應用程式取得用戶端 ID、用戶端密鑰、權杖網址和授權網址。
- 在「驗證供應商憑證」部分,輸入下列資訊:
- 用戶端 ID
- 用戶端密碼
- 權杖網址
- 授權網址
- 按一下「新增供應商設定」。
新建立的驗證供應商會顯示在「驗證供應商」清單中。
Google Cloud CLI
-
使用授權和權杖網址建立驗證提供者:
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" -
如要授予代理程式身分使用驗證供應商的權限,請更新專案或特定驗證供應商的 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"
連接器層級 (捲曲)
如要授予單一代理程式特定驗證供應商的存取權,請使用
setIamPolicyAPI。這項指令會覆寫資源上現有的任何允許政策。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 會傳回 LRO,其中中繼資料包含
auth_uri和consent_nonce。 - 前端重新導向:應用程式必須將使用者重新導向至
auth_uri。 - 完成:使用者授予同意聲明後,請使用
consent_nonce呼叫FinalizeCredential,完成流程並取得權杖。
更新用戶端應用程式
如要處理 3-legged OAuth 的使用者登入和重新導向,用戶端應用程式必須實作下列步驟,管理使用者同意聲明並繼續對話:
如需完整實作範例,請參閱「ValidateUserId 前端範例」。
處理授權觸發程序
如果代理程式需要使用者同意,就會傳回 adk_request_credential 函式呼叫。應用程式必須攔截這項呼叫,才能啟動使用者授權對話方塊或重新導向。
記錄授權供應商提供的 consent_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。 - 使用這些參數呼叫驗證供應商的
FinalizeCredentialsAPI。 - 收到成功回應後,請關閉授權視窗。
@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時,請務必啟用代理身分。
如要部署至 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]"],
},
)