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

如需代表特定最终用户向代理授予对外部工具和服务（例如 Jira 任务或 GitHub 代码库）的访问权限，请在 Agent Identity Auth Manager 中配置三方模式 OAuth 身份验证提供方。

通过管理凭据和令牌，三方模式 OAuth 身份验证提供方无需使用自定义代码来处理身份验证流程。

三方模式 OAuth 工作流程

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

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

准备工作

  1. 确认您已选择正确的身份验证方法。

  2. 启用 Agent Identity API。

    启用 API 所需的角色

    如需启用 API，您需要拥有 Service Usage Admin IAM 角色 (roles/serviceusage.serviceUsageAdmin)，该角色包含 serviceusage.services.enable 权限。了解如何授予角色。

    启用 API

  3. 创建和部署代理。
  4. 确保您有一个前端应用来处理用户登录提示和重定向到第三方同意页面。
  5. 验证您是否拥有完成此任务所需的角色。

所需的角色

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

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

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

所需权限

如需创建和使用三方身份验证提供程序，您需要具备以下权限：

  • 如需创建身份验证提供方，请执行以下操作： agentidentity.authProviders.create
  • 如需使用身份验证提供方，请执行以下操作：
    • agentidentity.authProviders.retrieveCredentials
    • aiplatform.endpoints.predict
    • aiplatform.sessions.create

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

创建三方身份验证提供方

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

如需创建三方身份验证提供方，请使用 gcloud CLI：

  1. 配置 OAuth 客户端应用以注册客户端并获取客户端 ID 和客户端密钥。使用相应部分中的模板指定重定向 URI。

  2. 使用客户端凭据创建身份验证提供方：

    gcloud alpha agent-identity authProviders create AUTH_PROVIDER_NAME \
    --project="PROJECT_ID" \
    --location="LOCATION" \
    --three-legged-oauth-client-id="CLIENT_ID" \
    --three-legged-oauth-client-secret="CLIENT_SECRET" \
    --three-legged-oauth-authorization-url="AUTHORIZATION_URL" \
    --three-legged-oauth-token-url="TOKEN_URL"
  3. 验证您的身份验证提供程序是否显示在列表中，以及其状态是否为 ENABLED： 
    gcloud alpha agent-identity authProviders list \
   --project="PROJECT_ID" \
   --location="LOCATION"

  4. 授予访问权限，以便您的代理和本地开发环境从身份验证提供方检索凭据。如需允许已部署的代理和您的个人用户账号访问身份验证提供方，请向身份验证提供方资源授予 Agent Identity User (roles/agentidentity.user) 角色：

    1. 向已部署代理的 SPIFFE ID（代理身份）授予访问权限：

      gcloud alpha agent-identity authProviders add-iam-policy-binding AUTH_PROVIDER_NAME \
    --project="PROJECT_ID" \
    --location="LOCATION" \
    --role="roles/agentidentity.user" \
    --member="principal://agents.global.org-ORGANIZATION_ID.system.id.goog/resources/aiplatform/projects/PROJECT_NUMBER/locations/LOCATION/reasoningEngines/ENGINE_ID"

    2. 为您的个人用户账号授予本地开发和测试权限 (adk web)：

      gcloud alpha agent-identity authProviders add-iam-policy-binding AUTH_PROVIDER_NAME \
    --project="PROJECT_ID" \
    --location="LOCATION" \
    --role="roles/agentidentity.user" \
    --member="user:USER_EMAIL"

替换以下内容：

  • PROJECT_ID：您的 Google Cloud 项目 ID。
  • LOCATION：身份验证提供方和代理的部署位置（例如 us-west1）。
  • AUTH_PROVIDER_NAME：身份验证提供方的名称（例如 bigquery-mcp-3lo-authprovider）。
  • AUTHORIZATION_URL：授权服务器网址（例如 https://accounts.google.com/o/oauth2/v2/auth）。
  • TOKEN_URL：令牌服务器网址（例如 https://oauth2.googleapis.com/token）。
  • CLIENT_ID：您从第三方服务生成的 OAuth 客户端 ID。
  • CLIENT_SECRET：您从第三方服务生成的 OAuth 客户端密钥。
  • ORGANIZATION_ID：您的 Google Cloud 组织 ID。
  • PROJECT_NUMBER：您的 Google Cloud 项目编号。
  • ENGINE_ID：已部署的推理引擎代理的 ID。
  • USER_EMAIL：您的个人用户账号电子邮件地址。

配置 OAuth 客户端应用

在注册 OAuth 客户端凭据之前，请从第三方授权服务器（例如 Google、GitHub 或 Jira）获取客户端 ID 和客户端密钥。

如果您要连接到Google Cloud之外的第三方服务，请从该服务的开发者门户获取 OAuth 客户端凭据，然后跳过本部分中的步骤。

注册重定向 URI

配置 OAuth 客户端凭据时，您必须注册身份验证提供方的专用回调重定向 URI。

  1. 使用以下模板构建重定向 URI：

    https://agentidentitycredentials.googleapis.com/v1alpha/projects/PROJECT_ID/locations/LOCATION/authProviders/AUTH_PROVIDER_NAME/oauthcallback

    替换以下内容：

    • PROJECT_ID：您的 Google Cloud 项目 ID。
    • LOCATION：身份验证提供程序将部署到的区域（例如 us-west1）。
    • AUTH_PROVIDER_NAME：您的身份验证提供程序的名称。

    例如：https://agentidentitycredentials.googleapis.com/v1alpha/projects/my-project/locations/us-west1/authProviders/bigquery-mcp-3lo-authprovider/oauthcallback。

  2. 如果您要连接到 Google Cloud 服务（例如 BigQuery），可以在 Google Cloud 控制台中配置同意屏幕并创建 OAuth 客户端凭据：

    1. 配置 OAuth 权限请求页面：

      1. 在 Google Cloud 控制台中，前往 API 和服务 > OAuth 权限请求页面。

        依次前往“API 和服务”>“OAuth 权限请求页面”

      2. 在应用信息部分中，输入应用名称（例如“BigQuery 管理器应用”）和支持电子邮件地址。
      3. 在受众部分，选择内部或外部。
      4. 输入您的联系信息以接收通知。
      5. 阅读并接受 Google API 服务用户数据政策。
      6. 点击完成。

    2. 创建 OAuth 客户端凭据：

      1. 在 Google Cloud 控制台中，依次前往 API 和服务 > OAuth 权限请求页面 > 客户端页面。

        依次前往“API 和服务”>“OAuth 权限请求页面”>“客户端”

      2. 点击创建凭据 >OAuth 客户端 ID。
      3. 从列表中选择 Web 应用选项。
      4. 为您的 OAuth 客户端输入一个易于识别的名称。
      5. 在已获授权的重定向 URI 部分中，点击添加 URI，然后输入您构建的重定向 URI。
      6. 点击创建。在已创建 OAuth 客户端对话框中，复制生成的客户端 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 Google Cloud auth provider
CredentialManager.register_auth_provider(GcpAuthProvider())

# The URI to redirect the user to after consent is granted.
CONTINUE_URI = "https://YOUR_FRONTEND_URL/validateUserId"

# Create auth provider scheme
# Note: If using the legacy V1 API, the resource name uses 'connectors'
# instead of 'authProviders': projects/.../connectors/...
auth_scheme = GcpAuthProviderScheme(
    name="projects/PROJECT_ID/locations/LOCATION/authProviders/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="AGENT_NAME",
    model="gemini-2.5-flash",
    instruction="AGENT_INSTRUCTIONS",
    tools=[toolset],
)

示例：连接到 BigQuery MCP

以下示例展示了将代理连接到 BigQuery MCP 服务器的 agent.py 配置：

import os
from google.adk.agents import Agent
from google.adk.apps import App
from google.adk.auth.credential_manager import CredentialManager
from google.adk.integrations.agent_identity import GcpAuthProvider, GcpAuthProviderScheme
from google.adk.models import Gemini
from google.adk.tools.mcp_tool.mcp_session_manager import StreamableHTTPConnectionParams
from google.adk.tools.mcp_tool.mcp_toolset import McpToolset
import google.auth
from google.genai import types

_, project_id = google.auth.default()
os.environ["GOOGLE_CLOUD_PROJECT"] = "PROJECT_ID"
os.environ["GOOGLE_GENAI_USE_VERTEXAI"] = "True"

bigquery_mcp_auth_provider_id = "AUTH_PROVIDER_NAME"
bigquery_mcp_endpoint = os.environ.get(
    "BIGQUERY_MCP_ENDPOINT", "https://bigquery.googleapis.com/mcp"
)

# Register Google Cloud auth provider
CredentialManager.register_auth_provider(GcpAuthProvider())

# URI to redirect user to after consent is granted.
CONTINUE_URI = "http://127.0.0.1:8501/validateUserId"

bigquery_mcp_auth_scheme = GcpAuthProviderScheme(
    name=f"projects/{project_id}/locations/LOCATION/authProviders/{bigquery_mcp_auth_provider_id}",
    scopes=["https://www.googleapis.com/auth/bigquery"],
    continue_uri=CONTINUE_URI,
)

bigquery_mcp_tools = McpToolset(
    connection_params=StreamableHTTPConnectionParams(url=bigquery_mcp_endpoint),
    auth_scheme=bigquery_mcp_auth_scheme,
    errlog=None,
)

root_agent = Agent(
    name="root_agent",
    model=Gemini(
        model="gemini-2.5-flash",
        retry_options=types.HttpRetryOptions(attempts=3),
    ),
    instruction=(
        "You are a helpful AI assistant designed to provide accurate and useful"
        " information. You can also use your BigQuery MCP tools to look up"
        " BigQuery data."
    ),
    tools=[bigquery_mcp_tools],
)

app = App(
    root_agent=root_agent,
    name="AGENT_NAME",
)

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/authProviders/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="AGENT_NAME",
    model="gemini-2.5-flash",
    instruction="AGENT_INSTRUCTIONS",
    tools=[spotify_search_track_tool],
)

app = App(
    name="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 scheme
auth_scheme = GcpAuthProviderScheme(
    name="projects/PROJECT_ID/locations/LOCATION/authProviders/AUTH_PROVIDER_NAME",
    continue_uri=CONTINUE_URI
)

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

toolset = registry.get_mcp_toolset(
    mcp_server_name=(
        "projects/PROJECT_ID/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="AGENT_NAME",
    model="MODEL_NAME",
    instruction="AGENT_INSTRUCTIONS",
    tools=[toolset],
)

直接调用 API

如果您未使用 ADK，您的代理必须调用 agentidentitycredentials.retrieveCredentials API 来获取令牌。

由于这是一个多步骤 OAuth 流程，因此您的代理必须处理操作的生命周期：

  1. 初始请求：代理调用 retrieveCredentials。
  2. 需要征得用户同意：如果用户未授予同意，API 会返回一个包含结果 uri_consent_required 的响应。这将包含 authorization_uri 和 consent_nonce。
  3. 前端重定向：您的应用必须将用户重定向到 authorization_uri。
  4. 完成：在用户表示同意后，使用 consent_nonce 调用 FinalizeCredentials 以完成流程并获取令牌。

更新客户端应用

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

示例界面服务器

您可以下载并运行使用 uvicorn 的完整示例界面服务器。开始之前，请确保您拥有 GitHub 账号并已安装 pip。

如需设置和运行示例界面服务器，请执行以下操作：

  1. 克隆 adk-python GitHub 代码库：

    git clone https://github.com/google/adk-python.git

  2. 前往代码库并激活 Python 虚拟环境：

    cd adk-python
python3 -m venv .venv
source .venv/bin/activate

  3. 前往示例界面客户端目录：

    cd contributing/samples/integrations/gcp_auth/client

  4. 安装客户端依赖项：

    pip install -r requirements.txt

  5. 在启动服务器之前，请设置 AGENT_PROJECT_DIR 环境变量以指定代理代码所在的目录。否则，应用默认会在客户端目录的父文件夹中查找代理。

    使用 uvicorn 启动示例界面服务器。确保端口与 OAuth 客户端中配置的重定向 URI 相匹配：

    export AGENT_PROJECT_DIR="/path/to/your/agent_project"
uvicorn main:app --port 8501 --reload

  6. 在浏览器中打开 http://localhost:8501。（注意：您必须使用 localhost，而不能使用 127.0.0.1，因为 OAuth 重定向网址专门要求使用 localhost 主机名。）指定设置，点击保存并应用设置，然后与代理互动。

自定义界面应用

如需在自定义界面应用中直接实现这些功能，请执行以下步骤：

处理授权触发器

当代理需要用户同意时，它会返回 adk_request_credential 函数调用。您的应用必须拦截此调用，以启动用户授权对话框或重定向。

通过记录身份验证提供方提供的 consent_nonce 来管理会话上下文。在验证步骤中，需要此随机数来验证用户。在会话中保存 auth_config 和 auth_request_function_call_id 值，以便在用户表示同意后轻松恢复流程。

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

实现用户验证端点

在您的 Web 服务器上实现验证端点（与配置期间提供的 continue_uri 相同的 URI）。此端点必须执行以下操作：

  1. 接收 user_id_validation_state、auth_provider_name 和 uuid 作为查询参数。
  2. 从会话存储空间中检索 user_id 和 consent_nonce 值。如果多个授权流程同时运行，请使用 uuid 来匹配正确的会话。
  3. 使用以下参数调用身份验证提供程序的 FinalizeCredentials API。
  4. 在收到成功响应后关闭授权窗口。
示例：FastAPI 验证端点 (main.py)

以下示例展示了一个完整的 FastAPI 验证端点，该端点可处理 OAuth 回调并最终确定用户凭据：

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

    base_url = "https://agentidentitycredentials.googleapis.com/v1alpha"
    finalize_url = f"{base_url}/{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时，请确保已启用代理身份。

如果您要部署到 Gemini Enterprise Agent Platform 上的 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]"
        ],
    },
)

后续步骤