Google ADK 智能体治理实战使用 agent-governance-toolkit 的 GoogleADKKernel 实施工具级策略执行【免费下载链接】agent-governance-toolkitAI Agent Governance Toolkit — Policy enforcement, zero-trust identity, execution sandboxing, and reliability engineering for autonomous AI agents. Covers 10/10 OWASP Agentic Top 10.项目地址: https://gitcode.com/GitHub_Trending/ag/agent-governance-toolkit导读本文将带你基于 examples/adk-governed 示例完整掌握如何为 Google ADKAgent Development Kit智能体接入 Agent Governance Toolkit 的治理能力包括工具白名单/黑名单、危险内容检测、人工审批、工具调用配额、预算控制与审计追踪。读完本文你将能够在不依赖运行中的 ADK 服务的前提下通过GoogleADKKernel回调机制为 ADK Agent 建立可审计、可阻断、可追溯的治理防线并理解其底层源码实现与生产级接入方式。一、背景为什么 ADK 智能体需要治理Google ADK 是多智能体开发框架支持LlmAgent、SequentialAgent、ParallelAgent、LoopAgent等形态。智能体在自主执行时会调用工具Tool完成搜索、发邮件、执行命令等操作而这些调用一旦失控可能带来破坏性后果——例如执行rm -rf、发送危险 SQL、超出预算或越权调用敏感工具。Agent Governance Toolkit 提供GoogleADKKernel适配器在 ADK 的工具调用、Agent 生命周期等钩子点注入治理逻辑。根据 integrations.md 的适配器清单GoogleADKKernel与 Anthropic、OpenAI Agents SDK、CrewAI 等 19 个框架适配器并列是项目官方支持的 Google ADK 集成入口。GoogleADKKernel的核心特性见 google_adk_adapter.py 模块文档通过 ADK 原生回调钩子before_tool_callback/after_tool_callback/before_agent_callback/after_agent_callback实施策略支持 Runner 级别的GovernancePlugin覆盖 ADK 全部 12 个生命周期钩子提供基于 blocked patterns 的内容过滤、工具白/黑名单、敏感工具人工审批、Token/调用预算跟踪、SIGKILL 取消运行支持完整的工具调用与 Agent 运行审计轨迹优雅导入处理未安装google-adk包时模块仍可正常加载_HAS_ADK False因此示例可以脱离真实 ADK 环境直接演示治理逻辑。二、环境准备与安装示例的依赖声明位于 requirements.txtagent-os-kernel安装方式pip install agent-os-kernel两个示例脚本通过sys.path.insert将agent-governance-python/agent-os/src加入模块搜索路径直接导入本地源码中的适配器_REPO_ROOT Path(__file__).resolve().parent.parent.parent sys.path.insert( 0, str(_REPO_ROOT / agent-governance-python / agent-os / src), ) from agent_os.integrations.google_adk_adapter import GoogleADKKernel注意示例特意避开需要额外 policy-engine 组件与原生绑定的高级 AGT v5 运行时路径因此可以在标准 Python 环境中直接运行无需编译或额外守护进程。三、快速上手getting_started.py 的最小治理演示getting_started.py 是一个最小化入门脚本演示五项核心能力工具白名单、工具黑名单、危险内容检测、治理违规记录、审计日志。运行方式python examples/adk-governed/getting_started.py3.1 创建治理内核脚本首先构造一个GoogleADKKernel实例kernel GoogleADKKernel( max_tool_calls10, allowed_tools[search, summarize], blocked_tools[execute_shell, run_command], blocked_patterns[DROP TABLE, rm -rf], on_violationlambda _e: None, )各参数的语义如下表参数示例值作用max_tool_calls10每次运行允许的最大工具调用次数超出后拒绝后续调用allowed_tools[search, summarize]工具白名单只有名单内的工具才被允许执行blocked_tools[execute_shell, run_command]工具黑名单名单内的工具一律拒绝blocked_patterns[DROP TABLE, rm -rf]危险内容模式工具参数中命中即被阻断require_human_approvalTrue演示脚本中是否要求敏感工具触发人工审批流程sensitive_tools[send_email]演示脚本中需要人工审批的敏感工具集合max_budget5.0演示脚本中单次运行的预算上限on_violationlambda _e: None违规回调可替换为日志、告警或上报逻辑说明白名单与黑名单是两条独立的策略路径——先命中黑名单直接拒绝不在白名单内的工具同样会被拒绝未授权工具两者共同构成工具级访问控制。3.2 四个演示场景脚本依次调用kernel.before_tool_callback(...)模拟 ADK 框架在工具执行前的钩子并根据返回值判断是否放行# [1] 尝试调用黑名单工具 execute_shell —— 被阻断 result kernel.before_tool_callback( tool_nameexecute_shell, tool_args{}, agent_nameresearch-agent, ) if result and result.get(error): print(f BLOCKED — {result[error]}) # [2] 尝试调用白名单外工具 web_scraper —— 被阻断 result kernel.before_tool_callback( tool_nameweb_scraper, tool_args{}, agent_nameresearch-agent, ) # [3] search 在白名单内但参数命中危险模式 DROP TABLE —— 被阻断 result kernel.before_tool_callback( tool_namesearch, tool_args{query: DROP TABLE users}, agent_nameresearch-agent, )返回值的契约非常清晰这一点在源码与测试中均有印证返回None允许执行返回{error: ...}字典拒绝执行error字段携带策略违规原因。3.3 审计统计与违规详情脚本末尾通过get_stats()与get_violations()汇总治理结果stats kernel.get_stats() violations kernel.get_violations() print(fViolations: {stats[violations]}) print(fAudit Events: {stats[audit_events]}) for i, violation in enumerate(violations, start1): print( f [{i}] {violation.policy_name}: f{violation.description} )从 google_adk_adapter.py 源码可见get_stats()返回的完整统计字段return { tool_calls: self._tool_call_count, agent_calls: self._agent_call_count, violations: len(self._violations), audit_events: len(self._audit_log), elapsed_seconds: round(time.time() - self._start_time, 2), }get_violations()返回PolicyViolationError列表每个违规对象暴露policy_name策略名与description违规描述等字段可直接用于报告或告警。完整的审计事件流则可通过get_audit_log()获取。四、端到端演示adk_governance_demo.py 的九大治理场景adk_governance_demo.py 在入门脚本基础上扩展为完整的端到端演示覆盖 9 个治理场景。运行方式python examples/adk-governed/adk_governance_demo.py其内核配置增加了人工审批与预算能力kernel GoogleADKKernel( max_tool_calls20, allowed_tools[search, summarize, send_email], blocked_tools[execute_shell, run_command], blocked_patterns[DROP TABLE, rm -rf], require_human_approvalTrue, sensitive_tools[send_email], max_budget5.0, on_violationlambda _e: None, )九个场景与预期结果#场景关键操作预期结果1Blocked Tool Enforcement调用execute_shell命中黑名单被阻断2Tool Allowlist Enforcement调用web_scraper不在白名单被阻断3Dangerous Content Detectionsearch参数含DROP TABLE users命中危险模式被阻断4Human Approval Workflow触发send_email进入人工审批流程5Sensitive Tool Approval再次触发send_email敏感工具走审批路径6Tool Call Limits用max_tool_calls3的内核连续调用 5 次超出配额后拒绝7Budget Controls查询预算上限打印budget_limit配置8Audit Trail Review读取get_stats()输出违规数与审计事件数9Governance Summary遍历get_violations()汇总全部违规详情其中场景 6 单独构造了一个低配额的受限内核来验证调用上限limit_kernel GoogleADKKernel( max_tool_calls3, blocked_tools[execute_shell], on_violationlambda _e: None, ) for i in range(5): result limit_kernel.before_tool_callback( tool_nameexecute_shell, tool_args{}, agent_nameresearch-agent, )场景 7 通过kernel.get_stats().get(budget_limit)打印预算配置场景 8/9 则复用get_stats()与get_violations()完成审计复盘。整个演示脚本的输出风格统一ALLOWED表示放行BLOCKED - 原因表示阻断便于直观观察每条策略的生效路径。五、策略清单adk-agt-manifest.yaml 解读仓库为 ADK 治理提供了示例策略清单 examples/adk-governed/policies/adk-agt-manifest.yaml完整内容如下# ADK Governance Policy — Sample Configuration version: 1.0 name: adk-governance description: Sample governance policy for Google ADK agents. Configures tool restrictions, rate limits, and delegation controls. disclaimer: This is a sample configuration. Customize for your environment. adk_governance: # Tools that are always blocked blocked_tools: - execute_shell - run_command - delete_database - drop_table # Maximum tool calls per agent per session max_tool_calls: 100 # Tools requiring human approval before execution require_approval_for: - send_email - publish_document - deploy_service - transfer_funds # Delegation controls delegation: max_depth: 3 require_scope_narrowing: true # Audit settings audit: log_all_tool_calls: true log_delegations: true include_tool_args: false # Set true only in dev (may contain PII)该清单展示了四类策略维度blocked_tools工具黑名单execute_shell、run_command、delete_database、drop_table等高危工具被无条件禁用max_tool_calls调用配额每个 Agent 每个会话最多 100 次工具调用require_approval_for审批工具send_email、publish_document、deploy_service、transfer_funds等敏感操作必须人工审批delegation委托控制委托深度上限为 3 层并要求委托时必须收窄作用域require_scope_narrowing: trueaudit审计设置记录所有工具调用与委托行为且默认不记录工具参数include_tool_args: false避免在生产环境审计日志中泄露 PII——只有在开发环境才建议开启。需要特别强调 README 中的说明该 YAML 文件是示例性illustrative配置。当前两个示例脚本为了简洁直接通过GoogleADKKernel构造函数配置治理规则并未加载这份 YAML。若要在生产环境中启用清单驱动的治理应走 AGT v5 的 manifest 运行时路径见下文第七节将策略统一收敛到配置文件中管理。六、源码级原理回调钩子的底层调用链6.1 内核架构从源码看GoogleADKKernel继承自BaseIntegration见 base.py内部持有_tool_call_count/_agent_call_count工具与 Agent 调用计数_audit_log: list[AuditEvent]审计轨迹AuditEvent记录时间戳、事件类型、Agent 名、详情以及 skill 溯源、上下文哈希等溯源字段google_adk_adapter.py_violations: list[PolicyViolationError]累计的违规记录_bridge: AdapterRuntime指向 AGT v5 的原生适配器运行时NativeAdapterRuntime所有策略判定最终委托给它_contexts: dict[str, ADKExecutionContext]按 invocation 维度跟踪运行状态。ADKExecutionContextgoogle_adk_adapter.py扩展了执行状态记录invocation_id、会话中出现的agent_names、运行历史、累计的prompt_tokens/completion_tokens、model_calls以及cancelled取消标志。6.2 before_tool_callback 的完整判定路径before_tool_callback是工具治理的核心入口google_adk_adapter.py其执行链如下提取上下文兼容 ADK 的ToolContext对象与 kwargs 两种传参方式分别取tool_name、tool_args、agent_name技能溯源审计通过emit_skill_audit_event发起POLICY_CHECK治理事件并记录before_tool审计条目委托运行时判定调用self._bridge.evaluate_pre_tool_call(...)将工具调用提交给原生 ACS 运行时评估得到AdapterResult含allowed、reason、transform、approval_required等属性见 _native_adapter_runtime.py拒绝处理若不允许调用_raise_violation(bridge_result)构造PolicyViolationError——先追加到_violations列表再触发on_violation回调最后返回{error: str(error)}阻断工具预算计数通过self._adapter_ctx.call_count 1更新调用计数供max_tool_calls预算策略读取。值得注意的细节AdapterResult.approval_required属性表明该拒绝是否可以由人工审批解除——当策略判定为 deny 且携带approval块时宿主端应引导走审批流程而非直接视为硬拒绝。6.3 输出侧治理after_tool_callback 与 after_agent_callback治理并不止于工具调用前。after_tool_callbackgoogle_adk_adapter.py在工具执行后对tool_result做输出侧检查将工具输出提交到evaluate_output干预点若策略不允许则返回{error: ...}阻断结果向 ADK 传播若策略携带transform重写/脱敏且结果类型匹配str或dict则用transformed_value就地替换工具输出——这正是脱敏策略落地的方式。before_agent_callback与after_agent_callback提供 Agent 生命周期侧的治理前者在 Agent 启动前将agent:{name}作为输入提交给运行时评估内容与审批门禁后者对最终输出做同样检查。6.4 Runner 级插件GovernancePlugin 与 12 个生命周期钩子对于生产环境推荐使用as_plugin()google_adk_adapter.py将内核包装为 ADK Runner 插件kernel GoogleADKKernel(blocked_tools[shell]) runner Runner( agentroot_agent, plugins[kernel.as_plugin()], )GovernancePlugingoogle_adk_adapter.py实现了 ADKBasePlugin的全部 12 个生命周期钩子on_user_message_callback对用户原始消息做内容过滤before_run_callback初始化ADKExecutionContext并检查取消状态before_agent_callbackAgent 调用限制与超时执行after_agent_callback输出内容过滤与审计before_model_callbackToken 预算预检与模型调用计数after_model_callback从LlmResponse提取 token 用量兼容 Gemini 与 LiteLLM/OpenAI 两种字段名on_model_error_callback记录模型错误before_tool_callback工具白/黑名单、内容扫描、人工审批after_tool_callback工具结果输出过滤on_tool_error_callback记录工具错误on_event_callback事件级审计富化after_run_callback运行结束审计摘要与上下文清理。插件回调在 Agent 级回调之前执行返回非 None 值即可短路执行流程。这样治理逻辑完整覆盖了从用户消息、模型调用、工具调用到运行结束的整个生命周期。6.5 运行时控制与健康检查内核还提供几组运维接口reset()重置计数器并轮换执行上下文使预算在下一轮运行中按新会话重新累计health_check()返回治理内核健康状态——存在违规时为degraded否则healthy并附带 ADK 可用性、模型调用数、Token 用量、被取消运行数等指标cancel_run(invocation_id)/is_cancelled(invocation_id)实现 SIGKILL 语义的运行取消——由于 ADK 运行是本地进程取消通过设置标志位实现所有治理钩子检测到后立即返回阻断响应。另外模块实现了graceful import当google-adk未安装时_HAS_ADK/_HAS_ADK_PLUGINS为False插件基类动态退化为object保证模块可导入、可测试——这也是本示例能在无 ADK 环境下演示的原因。七、测试印证与生产接入建议7.1 测试验证仓库测试对 ADK 回调行为做了显式钉定可作行为契约参考test_adapter_interception.py 中的test_google_adk_before_tool_callback_blocks_denied_tool验证对拒绝的工具before_tool_callback返回非 None 的阻断载荷并记录违规test_google_adk_before_tool_callback_allows_permitted_tool验证对允许的工具返回None放行test_provider_adapters.py 将GoogleADKKernel与其他 15 个框架适配器并列注册统一跑适配器契约测试agent-governance-python/agt-policies/tests/scenarios/test_google_adk_adapter_scenarios.py则从策略场景角度覆盖 ADK 适配器的行为。7.2 生产接入路径示例脚本展示的是直接回调调用direct callback invocation的轻量模式适合验证与单元级治理。生产环境推荐两条正式路径见 google_adk_adapter.py 的模块示例路径 A回调注入构造LlmAgent时通过**kernel.get_callbacks()注入四个回调钩子路径 BAgent 包装kernel.wrap(LlmAgent(...))包装 Agent 对象wrap()/get_callbacks()已标记废弃v1.0 将移除应迁移到插件方式路径 CRunner 插件推荐Runner(agentroot_agent, plugins[kernel.as_plugin()])充分利用 ADK 原生插件生命周期。同时v5 运行时的标准做法见 integrations.md是用 manifest 承载策略定义、工具目录、预算、转换与审批绑定from agent_control_specification import AgentControl runtime AgentControl.from_path(policies/manifest.yaml) kernel GoogleADKKernel(runtimeruntime)在这种模式下适配器构造器不再接受内联规则对象策略统一收敛到 manifest 中——示例 YAML adk-agt-manifest.yaml 可作为编写这类清单的起点。八、边界与注意事项示例不加载 YAMLgetting_started.py与adk_governance_demo.py通过内核构造函数直接配置治理规则仓库附带的策略清单仅作参考模板生产环境请走 manifest 驱动路径避免策略散落在代码中无需运行中的 ADK Agent示例通过直接调用before_tool_callback等回调模拟框架行为不要求启动 Google ADK 服务便于 CI 与本地验证依赖极简仅需agent-os-kernel规避了需要额外 policy-engine 组件和原生绑定的高级 v5 路径在标准 Python 环境即可运行审批语义拒绝结果可能携带approval_required标记表示该阻断可通过人工审批解除——接入方需实现对应的审批协调逻辑将审批结果反馈回策略引擎审计隐私示例清单默认include_tool_args: false生产审计日志默认不落工具参数防止 PII 泄露需要参数级审计时仅在开发环境开启。结语通过 examples/adk-governed 示例你可以在几分钟内为 Google ADK 智能体建立起完整的治理闭环工具白/黑名单控制调用面、危险模式拦截恶意参数、审批工作流约束敏感操作、调用配额与预算控制成本、审计日志与违规汇总保证可追溯。深入 google_adk_adapter.py 源码后可以发现这套能力建立在 ADK 原生回调钩子与 AGT 原生策略运行时之上——既有轻量直连的演示模式也有 Runner 插件与 manifest 驱动的生产路径可以随治理需求从验证平滑演进到规模落地。【免费下载链接】agent-governance-toolkitAI Agent Governance Toolkit — Policy enforcement, zero-trust identity, execution sandboxing, and reliability engineering for autonomous AI agents. Covers 10/10 OWASP Agentic Top 10.项目地址: https://gitcode.com/GitHub_Trending/ag/agent-governance-toolkit创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考