使用三方模式 OAuth 和身份验证管理器进行身份验证

如果您希望代理代表特定用户访问外部工具和服务（例如 Jira 任务或 GitHub 代码库），则必须在代理身份验证管理器中配置三方模式 OAuth 身份验证提供方。

三方模式 OAuth 身份验证提供方会为您管理用户重定向和令牌。这样一来，您就无需编写自定义代码来处理复杂 OAuth 2.0 流程。

三方模式 OAuth 工作流

三方模式 OAuth 身份验证提供方需要用户同意，因为代理会代表用户访问资源。

提示和重定向：聊天界面会提示用户登录然后将用户重定向到第三方应用的同意页面。
同意和存储：用户授予权限后，生成的 OAuth 令牌会存储在 Google 管理的凭据保险库中。
注入：当您使用智能体开发套件 (ADK) 时，代理会自动从身份验证提供方检索令牌，并将其注入到工具调用标头中。

准备工作

[验证您是否选择了正确的身份验证方法](/iam/docs/agent-identity-overview#auth-models)。
启用 Agent Identity Connector API。
启用 API 所需的角色
如需启用 API，您需要拥有 Service Usage Admin IAM 角色 (roles/serviceusage.serviceUsageAdmin)，该角色包含 serviceusage.services.enable 权限。了解如何授予角色。
启用 API
创建并部署代理。
确保您有一个前端应用来处理用户登录提示和重定向到第三方同意页面。
验证您是否拥有完成此任务所需的角色。

所需角色

如需获得创建和使用三方模式 Auth 提供方所需的权限，请让您的管理员为您授予项目的以下 IAM 角色：

如需创建身份验证提供方：
- IAM Connector Admin (roles/iamconnectors.admin)
- IAM Connector Editor (roles/iamconnectors.editor)
如需使用身份验证提供方：
- IAM Connector User (roles/iamconnectors.user)
- Agent Default Access (roles/aiplatform.agentDefaultAccess)
- Agent Context Editor (roles/aiplatform.agentContextEditor)
- Vertex AI User (roles/aiplatform.user)
- Service Usage Consumer (roles/serviceusage.serviceUsageConsumer)

如需详细了解如何授予角色，请参阅管理对项目、文件夹和组织的访问权限。

这些预定义角色包含创建和使用三方模式身份验证提供方所需的权限。如需查看所需的确切权限，请展开所需权限部分：

所需权限

创建和使用三方模式身份验证提供方需要以下权限：

如需创建身份验证提供方： iamconnectors.connectors.create
如需使用身份验证提供方：
- iamconnectors.connectors.retrieveCredentials
- aiplatform.endpoints.predict
- aiplatform.sessions.create

您也可以使用自定义角色或其他预定义角色来获取这些权限。

创建三方模式身份验证提供方

创建身份验证提供方，以定义第三方应用的配置和凭据。

如需创建三方模式身份验证提供方，请使用 Google Cloud 控制台或 Google Cloud CLI。

控制台

在 Google Cloud 控制台中，前往 Agent Registry 页面。
前往 Agent Registry
点击您要为其创建身份验证提供方的代理的名称。
点击身份。
在 Auth Providers 部分中，点击 Add auth provider 。
在 Add auth provider 窗格中，输入名称和说明。

名称只能包含小写字母、数字或连字符，不能以连字符结尾，并且必须以小写字母开头。
在 OAuth Type 列表中，选择 OAuth (3 legged) 。
点击创建并继续 。
如需授予代理身份使用身份验证提供方的权限，请点击授予访问权限 。
这会自动将 Connector User (roles/iamconnectors.user) 角色分配给身份验证提供方资源上的代理身份。
复制回调网址。
在单独的标签页中，在第三方应用上注册回调网址。
从第三方应用中，获取客户端 ID、客户端密钥、令牌网址和授权网址。
在 Auth provider credentials 部分中，输入以下信息：
- 客户端 ID
- 客户端密钥
- 令牌网址
- 授权网址
点击 Add provider config 。

新创建的身份验证提供方会显示在 Auth Providers 列表中。

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 允许政策，并将 Connector User (roles/iamconnectors.user) 角色授予代理正文。

代理身份基于行业标准的 SPIFFE ID 格式。在 IAM 允许政策中，代理身份使用主账号标识符 进行引用。

项目级 (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"

如需向项目中的所有代理授予对 Auth 提供方的访问权限，请运行以下命令：

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 会返回一个 LRO，其中元数据包含 auth_uri 和一个 consent_nonce。
前端重定向：您的应用必须将用户重定向到 auth_uri。
完成：用户授予同意后，调用 FinalizeCredential 使用 consent_nonce 以完成流程并获取令牌。

更新客户端应用

如需处理三方模式 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

实现用户验证端点

在 Web 服务器上实现验证端点（与配置期间作为 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时，请确保已启用 Agent Identity。

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