Google uses AI technology to translate content into your preferred language. AI translations can contain errors.

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

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

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

三方模式 OAuth 工作流程

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

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

准备工作

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

所需的角色

如需获得创建和使用三方身份验证提供程序所需的权限，请让您的管理员为您授予项目的以下 IAM 角色：

如需创建身份验证提供方，请执行以下操作：
- IAM Connector Admin (roles/iamconnectors.admin)
- IAM Connector Editor (roles/iamconnectors.editor)
如需使用身份验证提供方，请执行以下操作：
- IAM Connector User (roles/iamconnectors.user)
- 代理默认访问权限 (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

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

创建三方身份验证提供方

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

如需创建三方 Auth 提供方，请使用 Google Cloud 控制台或 Google Cloud CLI。

控制台

在 Google Cloud 控制台中，前往 Agent Registry 页面。
前往代理注册表
点击要为其创建身份验证提供程序的代理的名称。
点击身份。
在身份验证提供方部分中，点击 添加身份验证提供方。
在添加身份验证提供方窗格中，输入名称和说明。

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

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

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"
- 如需向项目中的所有代理授予对身份验证提供程序的访问权限，请运行以下命令：
  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 中的代理注册表 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 以完成流程并获取令牌。

更新客户端应用

为了处理用户登录和重定向（针对三方 OAuth），您的客户端应用必须实现以下步骤，以管理用户同意情况并恢复对话：

处理授权触发器
实现用户验证端点
继续与代理对话

如需查看完整的实现示例，请参阅验证 UserId 前端示例。

处理授权触发器

当代理需要用户同意时，它会返回 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

实现用户验证端点

在您的 Web 服务器上实现验证端点（在配置期间提供的相同 URI，即 continue_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 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]"],
    },
)