ARTICLE DETAIL

资讯详情

深耕郑州网站建设与运营推广的一线实战洞察。

ADK Python 工作流鉴权实战:用 FunctionNode 的 auth_config 实现 API Key 与 OAuth2 凭证收集

ADK Python 工作流鉴权实战:用 FunctionNode 的 auth_config 实现 API Key 与 OAuth2 凭证收集 ADK Python 工作流鉴权实战用 FunctionNode 的 auth_config 实现 API Key 与 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本文基于 Google ADK Python 仓库中的工作流鉴权示例 auth_api_key讲解如何在Workflow的FunctionNode上配置auth_config让节点在首次执行前自动挂起、向用户发起凭证收集adk_request_credential事件并在用户提交凭证后恢复运行。读完后你将掌握如何定义AuthConfig、如何在节点内通过ctx.get_auth_response()取出凭证、认证暂停/恢复的完整事件流以及同一套模式如何扩展到 OAuth2/OpenID Connect。一、机制总览认证暂停与恢复的四步流程当某个FunctionNode声明了auth_config时ADK 工作流引擎会自动执行以下流程原文档 Overview 章节所述节点暂停并向客户端发出一个名为adk_request_credential的 FunctionCall 事件本次调用invocation随之结束——该节点被标记为等待状态客户端携带凭证作为 FunctionResponse 发起新的请求工作流将凭证存入会话状态session state并重新运行该节点。ADK Web UIadk web会自动识别认证请求并弹出认证对话框无需任何额外代码。如果自研客户端则需要自行处理adk_request_credentialFunctionCall并用实际凭证作出响应。本示例采用API Key认证——这是最简单的一种凭证类型。示例中内置了一个模拟天气查询不需要任何外部 API 密钥或服务器当认证界面提示输入 key 时可以填入任意值如my-test-key-123。发送任意消息例如go即可启动整个工作流。二、示例目录结构与流程图示例位于 contributing/samples/workflows/auth_api_key/仅包含一个代理定义和一个测试会话文件作用agent.py定义auth_config、带认证的fetch_weather节点与summarize节点tests/go.json记录“发送 go → 触发认证 → 返回凭证 → 输出结果”的完整事件会话README.md本文所基于的说明文档工作流图如下fetch_weather在首次运行时会因认证而暂停用户提交 key 后重跑其输出流入summarize节点。三、第一步定义 AuthConfigAuthConfig是发给客户端的“凭证收集配置”定义了用什么认证方案收集凭证、用什么原始凭证进行后续交换、以及以什么键在凭证服务/会话状态中存取。完整定义见 auth_tool.py其核心字段为auth_scheme认证方案如APIKey、OAuth2、HTTP决定 UI 如何向用户索取凭证raw_auth_credential原始凭证。对于 OAuth2/OIDC 这类需要“交换”的方案它是交换的输入对 API Key 等方案可以为Noneexchanged_auth_credential交换后的凭证由 ADK 与客户端协作填充API Key 场景由客户端直接填写credential_key用户指定的存取键。若不显式提供源码会从 scheme/credential 的model_extra中提取credential_key否则回退到基于内容摘要的get_credential_key()auth_tool.py。API Key 场景的配置摘自 README.md 与 agent.pyfrom google.adk.auth.auth_tool import AuthConfig from google.adk.auth.auth_credential import AuthCredential, AuthCredentialTypes auth_config AuthConfig( auth_schemeAPIKey(**{in: APIKeyIn.header, name: X-Api-Key}), raw_auth_credentialAuthCredential( auth_typeAuthCredentialTypes.API_KEY, api_keyplaceholder, ), credential_keyweather_api_key, )说明APIKey/APIKeyIn来自fastapi.openapi.modelsin可取header/query/cookiename为传输凭证的字段名本例为请求头X-Api-Keyraw_auth_credential中的api_keyplaceholder只是占位真实值由用户在 UI 中提交credential_keyweather_api_key决定了该凭证在会话状态中的存取键。AuthCredential支持的凭证类型见 auth_credential.py 的AuthCredentialTypes枚举API_KEY、HTTP、OAUTH2、OPEN_ID_CONNECT、SERVICE_ACCOUNT。另外注意源码中的安全细节BaseModelWithConfig通过hide_input_in_errorsTrue和repr脱敏auth_credential.py防止密钥出现在校验错误信息与日志的 repr 输出中。四、第二步带认证门控的节点定义使用node装饰器传入auth_config并必须同时设置rerun_on_resumeTruenode(auth_configauth_config, rerun_on_resumeTrue) def fetch_weather(ctx: Context): cred ctx.get_auth_response(auth_config) api_key cred.api_key # Use api_key to call your API...这个约束在源码中是硬性校验的FunctionNode 在构造时检查if auth_config and not rerun_on_resume:直接抛出 “FunctionNode with auth_config requires rerun_on_resumeTrue.”。因为节点首次执行会被认证请求中断恢复后必须重跑整个函数体凭证才有机会被函数使用。完整示例中的节点与下游节点agent.py 修正为 contributing/samples/workflows/auth_api_key/agent.pynode(auth_configauth_config, rerun_on_resumeTrue) def fetch_weather(ctx: Context): Fetches weather data using the authenticated API key. # After auth completes, the credential is available via ctx. cred ctx.get_auth_response(auth_config) api_key cred.api_key if cred else unknown # In a real agent, you would use the api_key to call an external API. # For this sample, we just echo it back (masked). masked api_key[:4] **** if len(api_key) 4 else **** return { city: San Francisco, temperature: 18C, condition: Sunny, api_key_used: masked, } def summarize(node_input: dict): Displays the weather result. yield Event( message( fWeather for {node_input[city]}: f {node_input[temperature]}, {node_input[condition]}. f (Authenticated with key: {node_input[api_key_used]}) ) ) root_agent Workflow( nameauth_api_key, edges[(START, fetch_weather, summarize)], )summarize是一个无认证的普通函数节点它把上游fetch_weather的输出转成最终回复消息。五、认证门控的底层实现暂停、等待与重跑FunctionNode._run_impl开头的“Auth gate”是这套机制的核心src/google/adk/workflow/_function_node.py# --- Auth gate --- if self.auth_config: interrupt_id fwf_auth:{ctx.node_path} auth_response ctx.resume_inputs.get(interrupt_id) if auth_response is not None: await process_auth_resume( auth_response, self.auth_config, ctx.state, interrupt_id ) elif not has_auth_credential(self.auth_config, ctx.state): yield create_auth_request_event( self.auth_config, interrupt_id, ctx.state ) return从源码结构看其逻辑为以wf_auth:节点路径构造中断 ID若本次恢复请求的resume_inputs中携带了该中断 ID 的凭证响应则调用process_auth_resume把凭证写入ctx.state然后继续执行节点函数体若状态中还没有该auth_config对应的凭证has_auth_credential为 False则生成adk_request_credential事件并直接return——节点本轮不执行invocation 结束若凭证已存在于状态中例如同会话再次运行该节点则跳过请求直接执行。这正是 README 所述四步流程在代码层的对应暂停 → 等待 → 恢复 → 重跑。六、完整事件流从 go.json 测试会话看认证往返tests/go.json 完整记录了认证往返的事件序列可作为自研客户端实现时的参考用户事件e-1author: user内容为goinvocationId: i-1认证请求e-2author: auth_api_keycontent.parts中是一个functionCallname为adk_request_credential参数包含完整的authConfigauthScheme.inheader、nameX-Api-Key、credentialKeyweather_api_key等与提示语Please provide your API key for X-Api-Key.。同时该事件带有longRunningToolIds: [fc-1]nodeInfo.path为auth_api_key1/fetch_weather1——表明节点被挂起凭证响应e-3用户以functionResponse回传id对应fc-1response.result为 key 明文示例中为12345678节点输出e-4fetch_weather完成执行output为{city: San Francisco, temperature: 18C, condition: Sunny, api_key_used: 1234****}最终回复e-5summarize节点产出Weather for San Francisco: 18C, Sunny. (Authenticated with key: 1234****)。注意nodeInfo.path中1表示执行次数认证请求发出后节点重跑路径计数递增。自研客户端只需匹配第 2 步的adk_request_credentialFunctionCall弹出输入框收集 key再按第 3 步的格式回传functionResponse即可。七、在节点内读取凭证ctx.get_auth_responseContext.get_auth_response(auth_config)的实现在 context.py内部委托给AuthHandler(auth_config).get_auth_response(self.state)auth_handler.py即从会话状态中按credential_key读取认证完成后存储的AuthCredential读取不到时返回None。因此节点内推荐的防御式写法是cred ctx.get_auth_response(auth_config)后再判空示例中使用cred.api_key if cred else unknown。此外Context还提供了save_credential/load_credential经由credential_servicecontext.py用于跨会话持久化凭证而request_credential则面向 LLM Agent 工具调用场景。工作流节点场景下get_auth_response已足够。八、同一模式的 OAuth2 / OpenID Connect 扩展README 指出auth_config模式对 OAuth2 与 OpenID Connect 同样适用主要差异有四认证方案用OAuth2来自fastapi.openapi.models替代APIKey在 OAuth flows 中配置授权地址与令牌地址原始凭证设auth_typeAuthCredentialTypes.OAUTH2并在oauth2字段提供client_id、client_secret、redirect_uriWeb UI 行为ADK Web UI 识别 OAuth2 认证请求后自动打开授权弹窗用户向 IdP 完成认证UI 回传完整的AuthConfig响应节点内无需特殊处理令牌交换框架自动通过AuthHandler.exchange_auth_token()将授权码authorization code换取 access token。README 给出的完整示例from fastapi.openapi.models import OAuth2, OAuthFlowAuthorizationCode, OAuthFlows auth_config AuthConfig( auth_schemeOAuth2( flowsOAuthFlows( authorizationCodeOAuthFlowAuthorizationCode( authorizationUrlhttps://provider.com/authorize, tokenUrlhttps://provider.com/token, scopes{read: Read access}, ) ) ), raw_auth_credentialAuthCredential( auth_typeAuthCredentialTypes.OAUTH2, oauth2OAuth2Auth( client_idYOUR_CLIENT_ID, client_secretYOUR_CLIENT_SECRET, redirect_urihttp://localhost:8000/callback, ), ), credential_keymy_oauth_credential, )对照源码可以补充两点OAuth2Auth的完整字段包括auth_uri、state、auth_code、access_token、refresh_token、expires_in等auth_credential.py其中state用于把用户会话与授权请求绑定防止 CSRF 型混淆token_endpoint_auth_method默认为client_secret_basic也支持client_secret_post、client_secret_jwt、private_key_jwtAuthConfig.exchanged_auth_credential字段auth_tool.py的文档注释说明了 OAuth 流程中该字段的分工若原始凭证只有 client id/secretADK 负责生成授权 URI 与 state 并存入该字段客户端引导用户完成 OAuth 流程后把最终认证响应写回。九、验证与运行方式仓库为只读以下仅为查看与运行建议运行示例在仓库根目录安装依赖后将 contributing/samples/workflows/auth_api_key 作为 ADK 代理目录用adk web启动 Web UI需要fastapi及 ADK 的 workflow 相关依赖见 pyproject.toml输入go触发工作流首次运行fetch_weather时 UI 会弹出 API key 输入框提交任意值即可看到最终天气回复核对事件流直接阅读 tests/go.json它即上述第六节事件的权威参照可用于自研客户端的协议对齐深入认证子系统src/google/adk/auth/ 下还有auth_preprocessor.py、auth_schemes.py、credential_manager.py、exchanger/、refresher/、credential_service/等模块分别对应请求预处理、认证方案、凭证管理与刷新工作流侧的中断/恢复工具见 src/google/adk/workflow/utils/_workflow_hitl_utils.pycreate_auth_request_event、process_auth_resume、has_auth_credential均出自该模块。小结关注点结论触发条件FunctionNode设置auth_config且必须rerun_on_resumeTrue源码强校验暂停机制节点发出adk_request_credentialFunctionCall 后直接返回invocation 结束恢复机制客户端以 FunctionResponse 回传凭证框架写入 session state 并重跑节点节点内取凭证ctx.get_auth_response(auth_config)经AuthHandler从 state 读取免代码路径adk web自动处理 UI 弹窗与凭证回传扩展OAuth2/OIDC 换用OAuth2scheme OAuth2Auth原始凭证令牌交换由AuthHandler.exchange_auth_token()自动完成该示例的价值在于用最少的代码一个AuthConfig 一个node装饰器参数演示了 ADK 工作流“认证即中断点auth as an interruption point”的设计凭证收集对节点函数体完全透明节点代码只关心“拿到 key 后做什么”。【免费下载链接】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创作声明:本文部分内容由AI辅助生成(AIGC),仅供参考
返回列表