企业官网建设流程全解析

在 ADK-Python 中集成 GCP Agent Identity 认证：API Key、2LO 与 3LO 完整实战

【免费下载链接】adk-pythonAn open-source, code-first Python toolkit for building, evaluating, and deploying sophisticated AI agents with flexibility and control.项目地址: https://gitcode.com/GitHub_Trending/ad/adk-python

导读

本文基于 adk-python 仓库中的 GCP Auth 示例，完整讲解如何借助 Google Cloud Agent Identity 凭证服务（Credentials Service），为 ADK Agent 的工具接入 API Key、2-legged OAuth（2LO）与 3-legged OAuth（3LO）三种认证方式。示例以"查询 Spotify 曲目 + 读取私有播放列表 + 查询 Google Maps 天气"为业务场景，读者学完后可以掌握：GcpAuthProvider的注册方式、GcpAuthProviderScheme的配置要点、AuthenticatedFunctionTool与McpToolset的认证注入方法，以及使用adk web与自定义 FastAPI 客户端分别测试无交互认证和交互式授权流程的完整路径。

示例概览：一个 Agent 集成三类 GCP 认证

该示例位于 contributing/samples/integrations/gcp_auth，核心文件为 agent.py。它构建了一个名为gcp_auth的 ADK App，根 Agent 挂载了三个工具，分别对应三种不同的认证形态：

从源码结构可以推断，示例的测试路线也是按认证复杂度设计的：API Key 与 2LO 是无交互流程，可直接通过 ADK 官方 Web 客户端运行；3LO 涉及用户在浏览器中完成授权，因此配套提供了一个自定义 Web 客户端（client/main.py）来承载完整的授权回调与凭证落库流程。

环境准备：虚拟环境与依赖安装

激活虚拟环境

示例推荐使用独立的 Python 虚拟环境：

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

安装依赖

认证能力依赖agent-identity与mcp两个附加组件：

pip install "google-adk[agent-identity,mcp]"

其中agent-identity组件会引入google-cloud-agentidentitycredentials、google-cloud-iamconnectorcredentials等凭证服务客户端。从源码看，若缺少这些依赖，模块会直接抛出安装提示：

Missing required dependencies for Agent Identity Auth Manager. Please install with: pip install "google-adk[agent-identity]"

（见 _agent_identity_credentials_provider.py 与 _iam_connector_credentials_provider.py 的 ImportError 分支。）

认证本地环境：ADC 与配额项目

运行示例前，需要通过 Application Default Credentials（ADC）让本机身份具备访问凭证服务的权限：

gcloud auth application-default login export GOOGLE_CLOUD_PROJECT="YOUR_GOOGLE_CLOUD_PROJECT" gcloud auth application-default set-quota-project $GOOGLE_CLOUD_PROJECT export GOOGLE_GENAI_USE_ENTERPRISE=true

关键点说明：

• 运行 Agent 的身份（即 ADC 中的账号）必须拥有从这些 connector/provider 拉取凭证的 IAM 权限（iamconnectors.user一类角色），否则会在检索凭证时被拒绝；
• GOOGLE_GENAI_USE_ENTERPRISE=true用于启用企业版 Gemini 模型接入，示例 Agent 默认使用gemini-3.5-flash（见 agent.py）；
• 之后创建的 provider 资源名依赖GOOGLE_CLOUD_PROJECT与GOOGLE_CLOUD_LOCATION，agent.py 会从环境变量组装形如projects/{project}/locations/{location}/authProviders/{id}的资源全名。

创建 Auth Providers：三种 connector 的 gcloud 命令

参照 GCP 官方文档中关于管理 Auth Provider 的说明，在gcloud alpha agent-identity下创建三类 connector。请先导出公共参数：

export GOOGLE_CLOUD_LOCATION="YOUR_GOOGLE_CLOUD_LOCATION" export MAPS_API_AUTH_PROVIDER_ID="YOUR_MAPS_API_AUTH_PROVIDER_ID" export SPOTIFY_2LO_AUTH_PROVIDER_ID="YOUR_SPOTIFY_2LO_AUTH_PROVIDER_ID" export SPOTIFY_3LO_AUTH_PROVIDER_ID="YOUR_SPOTIFY_3LO_AUTH_PROVIDER_ID"

1. API Key 型 connector（Google Maps）

gcloud alpha agent-identity connectors create $MAPS_API_AUTH_PROVIDER_ID \ --project=$GOOGLE_CLOUD_PROJECT \ --location=$GOOGLE_CLOUD_LOCATION \ --api-key=YOUR_API_KEY

2. 2LO 型 connector（Spotify 服务端调用）

2LO 只需客户端凭据与令牌端点，不需要用户参与：

gcloud alpha agent-identity connectors create $SPOTIFY_2LO_AUTH_PROVIDER_ID \ --project=$GOOGLE_CLOUD_PROJECT \ --location=$GOOGLE_CLOUD_LOCATION \ --two-legged-oauth-client-id=OAUTH_CLIENT_ID \ --two-legged-oauth-client-secret=OAUTH_CLIENT_SECRET \ --two-legged-oauth-token-endpoint=OAUTH_TOKEN_ENDPOINT

3. 3LO 型 connector（Spotify 用户授权）

3LO 需要授权端点、令牌端点与允许的 scope：

gcloud alpha agent-identity connectors create $SPOTIFY_3LO_AUTH_PROVIDER_ID \ --project=$GOOGLE_CLOUD_PROJECT \ --location=$GOOGLE_CLOUD_LOCATION \ --three-legged-oauth-client-id=OAUTH_CLIENT_ID \ --three-legged-oauth-client-secret=OAUTH_CLIENT_SECRET \ --three-legged-oauth-authorization-url=AUTHORIZATION_URL \ --three-legged-oauth-token-url=TOKEN_URL \ --allowed-scopes=ALLOWED_SCOPES

注意：运行 Agent 的身份（ADC）必须具备从这些 connector 检索凭证的必要权限，请确保账号拥有相应角色后再继续。

理解 GcpAuthProvider 的底层路由

在深入示例代码前，先理解凭证提供者的实现，这对排查"为什么我的 provider 不生效"至关重要。

GcpAuthProvider 继承自BaseAuthProvider，其内部组合了两个底层实现：

• _IamConnectorCredentialsProvider：面向IAM Connector Credentials 服务（connector 资源）；
• _AgentIdentityCredentialsProvider：面向Agent Identity Credentials 服务（authProvider 资源）。

二者的选择依据是auth_scheme.name的正则匹配结果（gcp_auth_provider.py）：

if re.match(r"^projects/[^/]+/locations/[^/]+/connectors/[^/]+$", auth_scheme.name): return await self._iam_connector_provider.get_auth_credential(...) return await self._agent_identity_provider.get_auth_credential(...)

即：资源名含connectors段则走 IAM Connector 服务；否则按authProviders资源走 Agent Identity 服务。本示例中的MAPS_API_AUTH_PROVIDER、SPOTIFY_2LO_AUTH_PROVIDER、SPOTIFY_3LO_AUTH_PROVIDER均组装为.../authProviders/...形式，因此最终由 Agent Identity 凭证服务负责签发。

两个底层 Provider 都遵循相同的状态机（源码注释有明确说明）：

• API Key：一次请求即返回成功，直接构造HttpAuth凭证；
• 2LO：首次返回pending，需以 1 秒间隔轮询（NON_INTERACTIVE_TOKEN_POLL_INTERVAL_SEC = 1.0），最长 10 秒（NON_INTERACTIVE_TOKEN_POLL_TIMEOUT_SEC = 10.0）；
• 3LO：返回uri_consent_required，携带authorization_uri与consent_nonce，由客户端引导用户完成授权。

凭证构造时（见 _agent_identity_credentials_provider.py）：若服务返回的是Authorization: Bearer <token>头，则生成scheme="Bearer"的标准 HTTP 凭证；若是自定义头，则额外注入X-GOOG-API-KEY头，这正是API Key 自动注入的机制来源。

示例代码拆解：四种配置方式

完整代码见 agent.py，下面按 README 的步骤逐一拆解。

1. 注册 GCP Auth Provider

将GcpAuthProvider注册进全局CredentialManager，使其能够解析gcpAuthProviderScheme类型的认证配置。此操作只需执行一次：

CredentialManager.register_auth_provider(GcpAuthProvider())

对应实现位置为 agent.py。注册后，Agent 在运行工具时若遇到GcpAuthProviderScheme，CredentialManager会自动路由到该 Provider 拉取凭证。

2. 配置 2LO：AuthenticatedFunctionTool

先用GcpAuthProviderScheme指向 2LO connector 资源名，再把它包装进AuthConfig，最后附着到AuthenticatedFunctionTool：

spotify_auth_config_2lo = AuthConfig( auth_scheme=GcpAuthProviderScheme(name=SPOTIFY_2LO_AUTH_PROVIDER) ) spotify_search_track_tool = AuthenticatedFunctionTool( func=spotify_search_track, auth_config=spotify_auth_config_2lo, )

其中SPOTIFY_2LO_AUTH_PROVIDER为projects/{project}/locations/{location}/authProviders/{id}全名。2LO 场景不需要scopes与continue_uri，因为服务端到服务端的令牌交换由凭证服务代劳。

工具函数本身负责消费凭证。以spotify_search_track为例（agent.py）：它接收credential: AuthCredential参数，从credential.http中取出 token 组装Authorization头，再调用 Spotify 搜索接口：

async def spotify_search_track(credential: AuthCredential, query: str) -> str | list: headers = {} if http := credential.http: if http.scheme and http.credentials and (token := http.credentials.token): headers["Authorization"] = f"{http.scheme.title()} {token}" if http.additional_headers: headers.update(http.additional_headers) ...

AuthConfig定义于 auth_tool.py，核心字段为auth_scheme（认证方案）与exchanged_auth_credential（交换后的凭证，由 ADK 与客户端协作填充）。

3. 配置 3LO：交互式用户授权

3LO 需要用户授权，因此GcpAuthProviderScheme必须携带scopes与continue_uri：

spotify_auth_config_3lo = AuthConfig( auth_scheme=GcpAuthProviderScheme( name=SPOTIFY_3LO_AUTH_PROVIDER, scopes=["playlist-read-private"], continue_uri=CONTINUE_URI, ) ) spotify_get_playlist_tool = AuthenticatedFunctionTool( func=spotify_get_playlists, auth_config=spotify_auth_config_3lo, )

其中CONTINUE_URI = "http://localhost:8080/commit"（agent.py），它作为 OAuth 完成后的继续 URI：Google 托管的 OAuth 重定向 URI 会把用户重定向到这里，Agent 会在每次 3LO 请求中把该 URI 发给上游凭证服务。

从 GcpAuthProviderScheme 的定义可以确认其字段语义：

4. 配置 MCP Toolset 的自动认证

当使用McpToolset时，直接把auth_scheme传给工具集，MCP 服务器通信期间会自动完成认证（如 API Key 注入）：

maps_tools = McpToolset( connection_params=StreamableHTTPConnectionParams(url=MAPS_MCP_ENDPOINT), auth_scheme=GcpAuthProviderScheme(name=MAPS_API_AUTH_PROVIDER), errlog=None, # Required for agent-freezing (pickling) )

两点值得注意：

• MAPS_MCP_ENDPOINT = "https://mapstools.googleapis.com/mcp"（agent.py）是 Google Maps 工具集的 Streamable HTTP 端点；
• errlog=None是刻意为之：示例注释明确说明这是agent 冻结（pickling）所必需，避免日志对象在序列化时引发错误。

最后将三个工具挂到根 Agent 并包装为 App：

root_agent = Agent( name="gcp_auth_agent", model=MODEL, instruction=( "You are a Spotify and Google Maps assistant. Use your tools to " "search for track details, fetch the user's private playlists, " "and look up locations. ..." ), tools=[spotify_search_track_tool, spotify_get_playlist_tool, maps_tools], ) app = App(name="gcp_auth", root_agent=root_agent)

注意 README 中 Agent 名称为gcp_auth，而 agent.py 中根 Agent 的name为gcp_auth_agent，App 名为gcp_auth——在 ADK Web UI 中应选择名为gcp_auth的应用。

三组示例输入与对应的认证链路

README 给出了三组可直接验证的输入：

1. What is the current weather in New York?走 Google Maps 工具，验证API Keyauth provider：凭证服务直接返回密钥，注入X-GOOG-API-KEY头完成调用。

2. Tell me about the song: Waving Flag走 Spotify 搜索曲目工具，验证2LOauth provider：凭证服务先返回pending，客户端以 1 秒间隔轮询直至令牌签发，随后以Bearer头调用 Spotify API。

3. Get my private playlists走 Spotify 私有播放列表工具，验证3LOauth provider：必须使用自定义 Web 客户端完成浏览器授权，授权回调携带consent_nonce调用凭证服务的 Finalize 流程后才能获取令牌。

测试一：用adk web验证 API Key 与 2LO

API Key 与 2LO 均为无交互流程，直接启动 ADK Web 开发 UI 即可：

adk web contributing/samples/integrations

操作步骤：

1. 在 ADK Web UI 顶部的应用下拉框中选中gcp_auth；
2. 依次尝试上文第 1、2 组示例输入，分别验证 Maps（API Key）与 Spotify 搜索（2LO）是否正常返回。

adk web启动时会扫描指定目录下的 Agent 定义（包括 agent.py），因此传入的是其父目录contributing/samples/integrations。

测试二：用自定义 Web 客户端验证 3LO

3LO 需要完整的授权回调闭环，示例为此提供了 FastAPI 客户端（client/main.py）。

安装客户端依赖

cd contributing/samples/integrations/gcp_auth/client pip install -r requirements.txt

requirements.txt 中除google-adk[agent-identity,mcp]外，还包含fastapi、uvicorn、httpx、google-auth以及google-cloud-aiplatform[agent-engines]>=1.148.1（后者用于远程 Agent 引擎的发现与调用）。

启动客户端

uvicorn main:app --port 8080 --reload

然后打开http://localhost:8080。注意：必须使用localhost而不是127.0.0.1，因为 OAuth 重定向 URL 对此有严格要求（continue_uri为http://localhost:8080/commit）。

选择 Agent 类型并加载

客户端同时支持本地与远程 Agent（见 main.py 的ChatRequest模型）：

• Local Agent：从下拉框选择本地 Agent 模块（如agent）。客户端通过AGENT_PROJECT_DIR环境变量（默认取 gcp_auth 目录）定位模块，用InMemoryRunner运行并缓存会话（local_runners全局缓存保证多轮对话会话不丢失）；
• Remote Agent：填入 GCP Project ID 与 Location，点击 "Load Remote Agents" 后选择引擎。客户端通过vertexai.Client(...).agent_engines.list()枚举已部署的 Agent 引擎。

3LO 授权闭环的实现细节

理解 main.py 能帮你更透彻地明白 3LO 全流程：

1. Agent 调用受 3LO 保护的工具时，底层凭证服务返回uri_consent_required，ADK 随即发出名为adk_request_credential的函数调用；
2. 客户端在/chat的 SSE 事件流中侦测到该函数调用（main.py），解析出auth_uri、consent_nonce与auth_config，以popup_auth_uri等字段注入事件，前端弹出授权窗口；
3. 用户在 Google 托管的授权页完成后被重定向到/commit（即continue_uri）；
4. /commit处理器校验 cookie 中的user_id、consent_nonce与查询参数user_id_validation_state、auth_provider_name，通过AuthProviderCredentialsServiceClient.finalize_credentials落库凭证（main.py），并返回"Authorization successful"页面；
5. 前端把auth_config作为adk_request_credential的函数响应回传/chat（is_auth_resume=true），Agent 据此完成授权恢复，随后真正调用 Spotify 私有播放列表接口。

从源码可以推断，/commit中的auth_provider_name同时兼容connector_name参数，并会把路径中的/connectors/规范化为/authProviders/（main.py），以适配两类凭证服务的资源命名差异。

常见问题与注意事项

1. 权限不足：运行 Agent 的 ADC 身份必须拥有从 connector/authProvider 检索凭证的 IAM 角色，否则get_auth_credential会抛出RuntimeError: Failed to retrieve credential ...。
2. 依赖缺失：务必安装google-adk[agent-identity,mcp]，缺少时导入即报错并提示安装命令。
3. 回调地址必须用 localhost：3LO 的continue_uri绑定http://localhost:8080/commit，使用127.0.0.1会导致重定向校验失败。
4. errlog=None不能省略：McpToolset未显式设置errlog时可能导致 agent 冻结（pickling）失败。
5. 凭证轮询超时：2LO 令牌交换采用 1 秒间隔、10 秒超时的轮询策略（源码常量NON_INTERACTIVE_TOKEN_POLL_INTERVAL_SEC/NON_INTERACTIVE_TOKEN_POLL_TIMEOUT_SEC），服务端令牌签发较慢时需要适当评估超时窗口。
6. 不支持的状态会静默降级：两个底层 Provider 对无法识别的服务状态抛出ValueError而非RuntimeError，因为BaseLlmFlow._resolve_toolset_auth会捕获ValueError记录日志并继续执行，避免一次本可恢复的认证状态直接中断整个调用（见 _agent_identity_credentials_provider.py 的注释说明）。

参考资源

• 示例完整代码：contributing/samples/integrations/gcp_auth/agent.py
• 自定义 Web 客户端：contributing/samples/integrations/gcp_auth/client/main.py
• Provider 实现：gcp_auth_provider.py、gcp_auth_provider_scheme.py
• 凭证服务底层实现：agent_identity 模块 README、_agent_identity_credentials_provider.py、_iam_connector_credentials_provider.py
• 认证配置模型：auth_tool.py
• 认证工具包装：authenticated_function_tool.py

关于 2LO 与 3LO 的授权细节，可进一步查阅 GCP IAM 文档中"使用 2LO 认证"与"使用 3LO 认证"的章节，本文不再赘述。

【免费下载链接】adk-pythonAn open-source, code-first Python toolkit for building, evaluating, and deploying sophisticated AI agents with flexibility and control.项目地址: https://gitcode.com/GitHub_Trending/ad/adk-python