企业官网建设流程全解析

ADK Python 工具级认证实战：AuthConfig、AuthenticatedFunctionTool 与 OAuth2 暂停-恢复机制深度解析

【免费下载链接】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

导读

当 Agent 的工具需要代表终端用户调用第三方 API（读取日历、邮箱、网盘文档）时，ADK（Agent Development Kit，Python 实现）通过AuthConfig声明工具所需的认证方式，并在缺少凭据时暂停当前调用、发起授权请求、待用户完成授权后再恢复执行。本文以官方指南 docs/guides/auth/tool_auth/index.md 为主线，结合 src/google/adk/auth 目录下的源码实现，完整讲解认证中断机制的工作原理、OAuth2 授权码流的端到端接入方式、凭据存储策略与配置项细节，帮助你在自己的 Agent 中正确接入带认证的工具。

AuthConfig：声明"这个工具需要什么凭据"

一个读取用户日历或文档的工具，需要属于该用户的凭据。只有终端用户本人能授予这个凭据，而授予动作必然离开 Agent 本体：打开一个授权同意页面，再带着重定向地址回来。这个"离开-回来"的往返无法在单次工具调用内完成，因此 ADK 将它建模为一次中断（interruption）：

1. 工具声明它需要什么凭据，并返回一个占位结果；
2. 本次调用（invocation）带着凭据请求结束；
3. 应用层运行同意流程，带着答案发起新一轮运行；
4. ADK 重新执行那个一直在等待的工具调用。

三个核心类协作完成这一机制，定义在 src/google/adk/auth/auth_tool.py：

• AuthScheme描述 API 期望如何被认证。它是fastapi.openapi.models.SecurityScheme（APIKey、HTTPBase、OAuth2等）、OpenIdConnectWithConfig与CustomAuthScheme的联合类型（见 src/google/adk/auth/auth_schemes.py）。
• AuthCredential保存秘密本身。auth_type决定其形态（API_KEY、HTTP、OAUTH2、OPEN_ID_CONNECT、SERVICE_ACCOUNT，枚举定义见 auth_credential.py），对应的字段（api_key、http、oauth2、service_account）持有实际值。
• AuthConfig把前两者配对，并附上credential_key作为该凭据在存储中的键。

AuthenticatedFunctionTool、BaseAuthenticatedTool和McpTool都接收AuthConfig，并把凭据的获取委托给CredentialManager（credential_manager.py）；LLM 流程中的 auth 请求处理器负责暂停调用并在稍后恢复等待中的工具调用；BaseCredentialService负责在轮次之间记住凭据。

一个值得注意的安全细节：凭据类统一继承自BaseModelWithConfig，其__repr_args__会将来路不明的额外字段值替换为<redacted>，避免密钥泄漏进日志和传给 LLM 的错误字符串（auth_credential.py）。

快速开始：一个需要 OAuth2 access token 的完整 Agent

下面的完整示例构建了一个"文档 Agent"：它只有一个工具list_documents，需要 OAuth2 授权码流获取访问令牌。运行后程序会打印授权 URL、等待你粘贴落地后的重定向地址，然后完成最初的请求。

import asyncio from fastapi.openapi.models import OAuth2 from fastapi.openapi.models import OAuthFlowAuthorizationCode from fastapi.openapi.models import OAuthFlows from google.adk.agents import LlmAgent from google.adk.apps import App from google.adk.auth import AuthConfig from google.adk.auth import AuthCredential from google.adk.auth import AuthCredentialTypes from google.adk.auth import OAuth2Auth from google.adk.auth.credential_service.in_memory_credential_service import InMemoryCredentialService from google.adk.runners import Runner from google.adk.sessions import InMemorySessionService from google.adk.tools.authenticated_function_tool import AuthenticatedFunctionTool from google.genai import types auth_config = AuthConfig( auth_scheme=OAuth2( flows=OAuthFlows( authorizationCode=OAuthFlowAuthorizationCode( authorizationUrl="https://provider.example.com/authorize", tokenUrl="https://provider.example.com/token", scopes={"documents.read": "Read your documents"}, ) ) ), raw_auth_credential=AuthCredential( auth_type=AuthCredentialTypes.OAUTH2, oauth2=OAuth2Auth( client_id="YOUR_CLIENT_ID", client_secret="YOUR_CLIENT_SECRET", redirect_uri="http://localhost:8080/callback", ), ), credential_key="documents_api", ) def list_documents(folder: str, credential: AuthCredential) -> list[str]: """Lists the documents in a folder.""" access_token = credential.oauth2.access_token # Call the provider's API with access_token here. return [f"{folder}/report.pdf"] agent = LlmAgent( name="documents_agent", instruction="Use list_documents to answer questions about the user's files.", tools=[ AuthenticatedFunctionTool(func=list_documents, auth_config=auth_config) ], ) runner = Runner( app=App(name="documents_app", root_agent=agent), session_service=InMemorySessionService(), credential_service=InMemoryCredentialService(), ) async def main(): session = await runner.session_service.create_session( app_name="documents_app", user_id="user" ) message = types.Content( role="user", parts=[types.Part(text="What is in my reports folder?")] ) while True: auth_call = None async for event in runner.run_async( user_id="user", session_id=session.id, new_message=message ): for function_call in event.get_function_calls(): if function_call.name == "adk_request_credential": auth_call = function_call if event.content and event.content.parts: for part in event.content.parts: if part.text: print(part.text) if auth_call is None: break # The run paused. Send the user through consent and hand back the redirect. requested = auth_call.args["authConfig"] oauth2 = requested["exchangedAuthCredential"]["oauth2"] print("Open this URL:", oauth2["authUri"]) oauth2["authResponseUri"] = input("Paste the URL you landed on: ") response = types.Part.from_function_response( name="adk_request_credential", response=requested ) response.function_response.id = auth_call.id message = types.Content(role="user", parts=[response]) asyncio.run(main())

要点说明：

• credential参数由框架注入，对模型不可见——模型只会看到folder参数，不会接触敏感令牌。
• adk webUI 会替你完成同意步骤；上面这段循环是自定义客户端（不使用adk web）时的等价实现。注意list_documents的签名包含credential: AuthCredential，这正是AuthenticatedFunctionTool通过inspect.signature识别并注入的特殊参数（见 authenticated_function_tool.py），该参数会被追加到_ignore_params，从而从工具 schema 中隐藏（L66）。

工作原理：从声明到暂停再到恢复

声明工具需要凭据

两种方式让工具获得认证能力：

• AuthenticatedFunctionTool包装普通函数（如上面的list_documents）；
• BaseAuthenticatedTool是类式等价物：继承它并实现_run_async_impl，凭据会作为关键字参数传入（base_authenticated_tool.py）。

两者的执行流程完全一致（见 L81-L95）：

1. 先向CredentialManager请求凭据；
2. 若拿到凭据，直接执行真正的工具逻辑；
3. 若拿不到（如 OAuth2 只有 client id/secret，尚需用户授权），则调用request_credential并返回response_for_auth_required（默认字符串"Pending User Authorization."）作为占位结果，而不是执行你的代码。

工具也可以手动完成这一流程：在工具内部使用tool_context.request_credential与tool_context.get_auth_response（前者需要function_call_id，因此只能在工具内使用）；在 agent 回调中则使用save_credential与load_credential。

暂停与恢复的六步时序

整个流程可以用下面的时序图概括（原文档中的mermaid图）：

结合 credential_manager.py 与 auth_preprocessor.py 的源码，拆解为六个步骤：

1. 查询凭据：工具调用CredentialManager.get_auth_credential。若原始凭据已可直接使用（API key、HTTP 凭据这类简单凭据），直接原样返回，流程完全不暂停——源码中_is_credential_ready只对API_KEY与HTTP两类返回True（L356-L367）。否则依次检查凭据服务、会话状态中的 auth 响应，以及是否属于无需用户的 client-credentials 流程（_is_client_credentials_flow同时支持 OAuth2 的clientCredentialsflow 与 OIDC 声明了client_credentialsgrant 的情况，见 L468-L489）。对授权码流程且无任何存储时，返回None。

2. 发起授权请求：工具调用request_credential。AuthHandler.generate_auth_request（auth_handler.py）为 OAuth2 / OIDC 方案构建授权 URL，写入exchanged_auth_credential.oauth2.auth_uri，同时生成state；当code_challenge_method为"S256"时会生成 PKCEcode_verifier（generate_auth_uri使用 authlib 的OAuth2Session.create_authorization_url，见 L386-L427）。整个配置被暂存在event_actions.requested_auth_configs中，以正在等待的工具调用 id 为键。

3. 发出暂停事件：flow 为每个请求单独发出一个携带名为adk_request_credential的长运行函数调用（long-running function call）的事件，参数为functionCallId（等待中的工具调用 id）与authConfig（第 2 步的配置）。键是 camelCase，因为配置按别名（alias）序列化。随后 flow 结束本次 invocation——这就是"暂停"。

4. 引导用户授权：应用读取authConfig.exchangedAuthCredential.oauth2.authUri，把用户送去授权页，收集重定向结果。

5. 携带答案恢复：你以新运行发起恢复，其消息是一个用户角色的Content，内含名为adk_request_credential的FunctionResponse。response 是同一份配置，把答案填入exchangedAuthCredential：可以是authResponseUri（包含授权码的完整重定向 URL），也可以是直接可用的accessToken。

6. 存储并重放：下一次模型调用前，auth 请求处理器把响应匹配回请求，先在 OAuth2 / OIDC 场景下用授权码换取令牌，再把凭据以temp:<credential_key>为键写入会话状态，然后重新执行那个等待中的工具调用（_store_auth_and_collect_resume_targets与_AuthLlmRequestProcessor，见 auth_preprocessor.py）。

两个决定恢复能否成功的细节：

• FunctionResponse的 id 必须是adk_request_credential调用的 id，而不是等待中的工具调用 id；后者通过functionCallId单独传递（AuthToolArguments的两个字段正是function_call_id与auth_config，见 auth_tool.py）。
• 恢复消息必须是最近一个带内容（content）且作者为user的事件——处理器只扫描这一条事件（auth_preprocessor.py）。

恢复阶段还有一个安全设计：客户端会原样回传配置，因此处理器只从客户端响应中取"浏览器往返的结果"（exchanged_auth_credential），而 auth scheme、raw credential、credential key 全部以服务端发出的请求为准（L158-L173）；AuthHandler在把配置发给客户端前会剥离 client secret（_without_client_secret），令牌交换时再从工具自身配置中重新挂载（auth_handler.py）。

凭据存储在哪里：临时状态与凭据服务

第 6 步写入的是带temp:前缀的状态键。临时状态按设计就是临时的：会话服务只在当前 invocation 内保留它，不做持久化。它单独只能解除等待中的工具调用，因此下一轮会再次要求用户授权。

让授权"持久生效"的是凭据服务（credential service），把它传给 Runner 即可（如快速开始示例中的InMemoryCredentialService）。此后CredentialManager会把交换后的凭据保存在credential_key下，后续调用时重新加载，并且会在 OAuth2 token 过期时刷新而非再次弹窗（_refresh_credential通过CredentialRefresherRegistry查找刷新器，见 credential_manager.py）。

仓库中现成的两类凭据服务：

• InMemoryCredentialService：进程内字典存储，按app_name -> user_id -> credential_key三层分桶（L58-L68），适合开发与单进程部署；
• SessionStateCredentialService：把凭据放进会话状态，注意其 docstring 明确警告"存于会话可能不安全，风险自负"。

两者都继承自 base_credential_service.py 中的BaseCredentialService。此外，CredentialManager内部默认注册了 OAuth2/OIDC 的OAuth2CredentialExchanger与OAuth2CredentialRefresher，以及 SERVICE_ACCOUNT 的ServiceAccountCredentialExchanger（L133-L157），你也可以用register_credential_exchanger/register_credential_refresher注册自定义实现。

配置选项速查

原文档给出的AuthConfig字段说明如下：

与源码对照可以补充三点：

• raw_auth_credential的校验：_validate_credential会强制要求 OAuth2 / OIDC scheme 必须有raw_auth_credential，且其oauth2子对象不能为空，否则抛出ValueError（L369-L401）。若 scheme 是带issuer_url的ExtendedOAuth2且缺少端点信息，还会尝试通过OAuth2DiscoveryManager自动发现 authorization/token 端点（L414-L447）。
• OAuth2Auth的可用字段（见 auth_credential.py）：除 client id/secret 外，还包括auth_uri、state、nonce、redirect_uri、auth_response_uri、auth_code、access_token、refresh_token、id_token、expires_at、expires_in、audience、prompt、code_verifier、code_challenge_method，以及默认值为client_secret_basic的token_endpoint_auth_method。其中秘密类字段（secret、token、code 等）均标记了repr=False，不会出现在 repr 输出中。
• credential_key的推导：AuthConfig.__init__优先使用显式传入的credential_key，否则从 raw credential / scheme 的model_extra中读取credential_key或credentialKey，最后才调用get_credential_key()生成adk_<scheme>_<credential>形式的摘要键（auth_tool.py）。注意get_credential_key()已被标记为 deprecated，官方建议直接设置credential_key。

已知限制

• 实验性 API：AuthenticatedFunctionTool、BaseAuthenticatedTool、CredentialManager、各凭据服务与凭据交换器均为实验特性。它们默认开启，首次使用时输出一次告警，但 API 可能变化。
• OAuth2 辅助功能依赖authlib：没有authlib时不会生成授权 URL、也不会用授权码换取令牌；凭据原样透传，客户端必须自己完成 OAuth 流程（AUTHLIB_AVAILABLE的判定见 auth_handler.py，缺失时generate_auth_uri直接返回原始凭据副本，见 L328-L333）。
• 会话状态不是秘密存储：SessionStateCredentialService会把令牌放进会话状态所在的位置。
• AuthConfig.get_credential_key()已弃用：请设置credential_key。

相关示例

• OAuth with the Calendar API：日历 API 的 OAuth2 授权码流完整接入示例。
• OAuth2 client credentials：client-credentials 流程——无需用户参与的机器对机器认证。
• MCP toolset auth, with the resume loop：MCP 工具集认证，包含与本文一致的恢复循环实现。
• API key auth on a workflow node：在工作流节点上使用 API key 认证。

小结

ADK 的工具级认证把"离开 Agent 完成用户授权"这一外部往返建模为一次可恢复的中断：AuthConfig声明需求，CredentialManager负责加载、交换、刷新与缓存，adk_request_credential长运行函数调用携带授权请求暂停调用，auth 请求处理器在下一轮恢复时把凭据存入临时状态（配合凭据服务实现跨轮持久），最后重新执行等待中的工具。理解这套"暂停-恢复"循环，是接入 OAuth2、OIDC 等用户授权类工具认证的关键；对 API key 与 HTTP 这类简单凭据，则无需任何暂停，配置即可直通使用。

【免费下载链接】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