Agent Governance Toolkit 协议面Protocol Facets面向线级语义的 SQL 与 Kubernetes 策略评估指南【免费下载链接】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 Agent 在执行数据库查询、Kubernetes API 调用等高危操作时传统策略引擎只能看到工具名或 HTTP 元数据无法识别一条 SQL 到底是SELECT还是DROP、一个 K8s 请求是否指向production命名空间。Agent Governance Toolkit 的Protocol Facets协议面机制在PolicyEngine.evaluate()进入规则匹配之前从原始协议上下文SQL 语句、Kubernetes API 路径中结构化提取sql.*与k8s.*字段并合并进评估上下文使 YAML 规则可以用点号dot-notation条件直接引用线级语义。阅读本文后你将掌握 SQL/K8s 面字段的提取规则、HTTP 方法到 Kubernetes 动词的映射、无侵入的 MCP 透明代理集成方式、自定义协议解析器的注册方法以及该模型在 Python、Rust、TypeScript、.NET 等多语言 SDK 中的一致性实现。工作机制规则求值前的结构化提取PolicyEngine.evaluate()的入口逻辑位于 agent-governance-python/agent-mesh/src/agentmesh/governance/policy.py# Populate sql.* and k8s.* fields before rules run from agentmesh.governance.protocol_facets import extract_protocol_facets extract_protocol_facets(context)也就是说每次调用evaluate()时系统都会先运行extract_protocol_facets(context)再评估规则。只要上下文context中包含sql或k8s子字典对应的解析器就会填充结构化字段随后 YAML 规则即可用{field, operator, value}条件引用这些字段。面提取发生在规则匹配之前因此规则作者完全不需要关心原始 SQL 解析或 K8s 路径正则的细节——只需按字段名书写条件。从源码结构看面提取模块 protocol_facets.py 由三部分组成FacetRegistry类按上下文键名持有面提取器提取器收到对应键的子字典并返回要合并的字段提取器内部异常会被捕获并记录日志单个解析器故障不会阻塞策略评估SQL 解析器_extract_sql_facets与 K8s 解析器_extract_k8s_facets模块级default_registry预注册了sql与k8s两个提取器并对外暴露extract_protocol_facets(context, registryNone)便捷函数未传 registry 时使用default_registry。SQL 面sql.*字段定义调用方在上下文中放入context[sql][query]SQL 语句字符串解析器提取以下字段供策略条件引用字段示例值说明sql.verbSELECT、DROP、DELETE大写形式的 SQL 动词sql.targetusers操作的主要表/对象语句涉及的第一张表sql.tablesorders,users引用的全部表的逗号连接列表sql.functionsCOUNT,NOW使用的 SQL 函数名的逗号连接列表依赖说明SQL 解析依赖sqlglotpip install sqlglot。未安装或解析失败时sql.verb会被置为UNKNOWN——这是一种fail-closed失败即拒绝倾向行为避免因解析器缺失而放过危险语句。底层解析原理从 protocol_facets.py 的源码可以看到_extract_sql_facets的实现思路通过sqlglot.parse(query)将语句解析为 AST再依据 AST 节点类型exp.Select、exp.Insert、exp.Update、exp.Delete、exp.Drop、exp.Create、exp.AlterTable、exp.Grant、exp.Merge等判定动词对于无法归类的Command节点通过_SQL_VERB_MAP按命令名映射覆盖TRUNCATE、REVOKE、CALL、EXECUTE、EXPLAIN、WITH等tables通过stmt.find_all(exp.Table)收集functions通过stmt.find_all(exp.Func)收集并转为大写target取语句涉及的第一张表例如SELECT * FROM orders JOIN users ...的 target 是orders。空查询或纯空白查询返回空字符串字段缺少sqlglot时返回UNKNOWN。这些边界行为在 test_protocol_facets.py 中有完整的测试覆盖包括test_no_sqlglot_returns_unknownmock 掉sqlglot导入后断言 verb 为UNKNOWN。示例规则每条规则通过{field, operator, value}匹配单个提取字段。复合检查例如动词 目标组合应拆分为多条带合适优先级的规则rules: - name: deny-destructive-sql condition: {field: sql.verb, operator: in, value: [DROP, TRUNCATE, DELETE]} action: deny priority: 100 - name: deny-schema-changes condition: {field: sql.verb, operator: in, value: [ALTER, GRANT, REVOKE]} action: deny priority: 100 - name: allow-read-only-sql condition: {field: sql.verb, operator: eq, value: SELECT} action: allow priority: 5示例评估上下文engine.evaluate( agent_diddid:example:agent1, context{sql: {query: DROP TABLE production}}, )上面的上下文经extract_protocol_facets处理后sql.verb为DROP、sql.target为production从而命中deny-destructive-sql规则。集成测试 TestPolicyEngineIntegration.test_sql_drop_denied_by_rule 验证了DROP TABLE users会被拒绝且matched_rule deny-drop同文件还验证了DROP TABLE staging非 protected 目标在带default_action: allow的策略下可以放行说明规则可以精确到“目标表”粒度。Kubernetes 面k8s.*字段定义调用方在上下文中放入context[k8s][method]HTTP 方法与context[k8s][path]API Server 路径解析器提取以下字段字段示例值说明k8s.verbget、list、delete、createKubernetes API 动词k8s.resourcepods、deployments资源类型k8s.namespaceproduction命名空间集群级资源为空k8s.namemypod资源名称集合请求为空k8s.subresourceexec、log子资源无则为空HTTP 方法到 Kubernetes 动词的映射HTTP具名资源集合GETgetlistDELETEdeletedeletecollectionPOSTcreatecreatePUTupdateupdatePATCHpatchpatch源码中还额外处理了HEAD具名资源映射为get集合映射为list。从 protocol_facets.py 可以看到判定“具名”还是“集合”的关键在于路径是否包含资源名称_K8S_PATH_PATTERNS中按“最具体优先”排列了 10 条正则覆盖/api/version/...与/apis/group/version/...两类前缀下的具名/集合/命名空间/子资源组合若路径匹配到name组则用_METHOD_TO_VERB_NAMED映射否则用_METHOD_TO_VERB_COLLECTION映射。例如GET /api/v1/namespaces/default/pods/mypod→ verb 为get而GET /api/v1/namespaces/default/pods→ verb 为list见 test_get_named_resource 与 test_get_collection。示例规则rules: - name: deny-k8s-production-namespace condition: {field: k8s.namespace, operator: eq, value: production} action: deny priority: 110 - name: deny-k8s-exec condition: {field: k8s.subresource, operator: eq, value: exec} action: deny priority: 100 - name: deny-k8s-deletecollection condition: {field: k8s.verb, operator: eq, value: deletecollection} action: deny priority: 100 - name: allow-k8s-readonly condition: {field: k8s.verb, operator: in, value: [get, list, watch]} action: allow priority: 5示例评估上下文engine.evaluate( agent_diddid:example:agent1, context{ k8s: { method: DELETE, path: /api/v1/namespaces/production/pods/mypod, } }, )该上下文解析后k8s.verb delete、k8s.namespace production、k8s.resource pods、k8s.name mypod命中deny-k8s-production-namespace。对应集成测试 test_k8s_delete_prod_denied 验证了 production 下的删除被拒绝而 test_k8s_delete_staging_allowed 验证了 staging 下的同形态删除在default_action: allow下放行——这是“按命名空间隔离风险”的典型用法。透明代理集成零应用改造MCP 代理agentmesh proxy会自动从工具调用参数中填充线级协议上下文无需修改任何应用代码名为query或sql的工具参数 → 填充context[sql][query]名为method/http_method配合path/api_path且路径以/api/或/apis/开头 → 填充context[k8s]因此只要在策略文件中定义 SQL 或 K8s 规则这些规则就会自动作用于所有经过代理的工具调用。这对于治理 Agent 通过 MCP 工具触发的数据库操作和集群管理操作非常实用治理规则与业务代码解耦新增协议面规则不需要任何应用侧改动。扩展自定义协议解析器Python注册到 default_registry面提取模块的文档字符串与测试用例test_register_and_extract_custom_protocol均演示了自定义协议的注册方式from agentmesh.governance.protocol_facets import default_registry def extract_redis_facets(redis_ctx: dict) - dict: cmd (redis_ctx.get(command) or ).upper() return {verb: cmd, key: redis_ctx.get(key, )} default_registry.register(redis, extract_redis_facets)之后在评估上下文中传入{redis: {command: FLUSHALL}}即可编写如下规则- name: deny-redis-flush condition: {field: redis.verb, operator: in, value: [FLUSHALL, FLUSHDB]} action: deny值得注意的几个设计细节均有测试佐证提取器接收的是子字典redis_ctx返回的字段会update回该子字典见 FacetRegistry.extract若上下文键的值不是字典例如sql: not-a-dict提取器会被跳过test_non_dict_context_key_skipped提取器抛出的异常会被捕获并记录日志不会阻断策略评估test_extractor_exception_is_swallowed多个提取器按注册顺序依次执行test_multiple_extractors_all_run_in_order也可以构造独立FacetRegistry并通过extract_protocol_facets(ctx, registry...)传入实现不同调用方使用不同解析集合test_custom_registry_passed_to_extract_protocol_facets。完整的规则示例参见 examples/policy-templates/wire-protocol-rules.yamlPython SDK 的agent_control_specification_version: 0.4.0-alpha.1格式策略模板。多语言一致性同一套面模型跨 SDK 可用同一面模型在多个语言 SDK 中保持一致每个 SDK 都暴露FacetRegistry、默认注册表以及extract_protocol_facets等价辅助函数内置sql.*与k8s.*提取器且字段名一致并在策略求值内部自动运行提取器。语言模块 / 包状态Pythonagentmesh.governance.protocol_facets已发布Rustagentmesh::protocol_facets已发布TypeScriptmicrosoft/agent-governance-sdk→protocol-facets仓库内已实现.NETagent-governance-dotnet→AgentGovernance.Policy.ProtocolFacets已实现Goagent-governance-golang已实现说明原文档以 GitHub issue 编号#2553、#2587 等标注各语言状态本仓库中 TypeScript 的实现在 agent-governance-typescript/src/protocol-facets.ts 及其测试 tests/protocol-facets.test.ts并通过 src/index.ts 对外导出Go 与 .NET 的规则示例文件分别位于 agent-governance-golang/examples/wire-protocol-rules.yaml 与 agent-governance-dotnet/examples/Quickstart/wire-protocol-rules.yaml。Rust 用法Rust SDK 在agentmesh::protocol_facets模块下提供同样的面模型FacetRegistry、default_registry()、extract_protocol_facets、extract_sql_facets、extract_k8s_facets并暴露相同的sql.*与k8s.*字段。PolicyEngine::evaluate会在调用方上下文的防御性副本上运行注册表再匹配规则因此不会修改调用方数据。use agentmesh::{PolicyEngine, default_registry}; use serde_yaml::Value; use std::collections::HashMap; let engine PolicyEngine::new(); engine.load_from_yaml(r# version: 1 agent: did:example:agent1 policies: - name: deny-destructive-sql type: capability denied_actions: [*] conditions: sql.verb: [DROP, TRUNCATE, DELETE] #).unwrap(); let mut sub serde_yaml::Mapping::new(); sub.insert(Value::String(query.into()), Value::String(DROP TABLE production.into())); let mut ctx HashMap::new(); ctx.insert(sql.to_string(), Value::Mapping(sub)); let decision engine.evaluate(db.exec, Some(ctx)); // decision PolicyDecision::Deny(...) // Register a custom protocol extractor: default_registry().register(redis, |sub| { let mut m std::collections::HashMap::new(); if let Some(cmd) sub.get(Value::String(command.into())).and_then(|v| v.as_str()) { m.insert(verb.to_string(), Value::String(cmd.to_uppercase())); } m });Rust SDK 的规则条件沿用现有 YAML 映射形态key: value或key: [v1, v2]表示in式成员判断字段名与决策结果与 Python 实现完全一致。完整的 Rust 规则示例见 agent-governance-rust/agentmesh/examples/wire-protocol-rules.yaml。SQL 解析器说明Rust 提取器使用内置的正则分词器覆盖策略规则常用的动词/目标/函数场景。对于复杂的方言级 SQL请通过default_registry().register(sql, ...)注册自定义提取器。.NET 用法.NET SDK 在AgentGovernance.Policy命名空间下暴露同样的面模型FacetRegistry、ProtocolFacets.DefaultRegistry、ProtocolFacets.ExtractProtocolFacets、ProtocolFacets.ExtractSqlFacets、ProtocolFacets.ExtractK8sFacets。PolicyEngine.Evaluate在内部上下文副本上运行注册表调用方只需填充原始sql/k8s子字典——调用方自己的字典不会被修改。using AgentGovernance.Policy; var engine new PolicyEngine(); engine.LoadYaml( apiVersion: governance.toolkit/v1 name: sql-guard scope: global default_action: allow rules: - name: deny-destructive-sql condition: sql.verb DROP action: deny priority: 100 ); var decision engine.Evaluate(did:mesh:agent1, new Dictionarystring, object { [sql] new Dictionarystring, object { [query] DROP TABLE production }, }); // decision.Allowed false, decision.MatchedRule deny-destructive-sql // Register a custom protocol extractor: ProtocolFacets.DefaultRegistry.Register(redis, sub { var cmd sub.TryGetValue(command, out var v) ? v?.ToString() ?? : ; return new Dictionarystring, object { [verb] cmd.ToUpperInvariant() }; });.NET 的规则条件使用表达式字符串格式sql.verb DROP、点路径字段引用、and/or组合字段名与高层决策结果与 Python 实现一致存在少量规则语法差异——尤其是 .NET 的in运算符引用的是列表值上下文字段而非 YAML 字面量列表因此多动词规则建议拆分为单独的检查参见 agent-governance-dotnet/examples/Quickstart/wire-protocol-rules.yaml 中的拆分写法如deny-destructive-sql-drop、deny-destructive-sql-truncate、deny-destructive-sql-delete各自独立成条。SQL 解析器说明.NET 提取器同样使用内置正则分词器并非完整的 SQL 解析器无法覆盖所有方言构造在高保障环境中建议通过ProtocolFacets.DefaultRegistry.Register(sql, ...)注册基于真正 SQL 解析器的自定义提取器。实战建议与注意事项fail-closed 语义SQL 解析在缺少sqlglot或解析失败时返回UNKNOWN动词。治理生产环境的 Agent 时应把UNKNOWN纳入 deny 规则或依赖default_action: deny避免“解析不出就放行”的漏洞。复合条件用多条规则表达Python 的{field, operator, value}每条规则匹配单个字段动词 目标的复合检查应拆分为多条规则并用优先级排序Rust 用conditions映射、.NET 用表达式字符串语法形态各有差异跨语言迁移时留意 各语言规则示例 的差异。面提取不会污染调用方数据Python 端extract_protocol_facets就地更新上下文并返回同一字典对象test_returns_same_dictRust 与 .NET 则在内部副本上运行——调用方无需担心原始上下文被修改或重复提取产生副作用。代理层自动生效通过agentmesh proxy透传的 MCP 工具调用只要参数命名符合query/sql与method/path/api/、/apis/前缀约定规则即可零改造生效自定义协议可遵循同样的参数命名约定并注册对应提取器。总结Wire-Protocol-Aware Policy Evaluation 通过协议面机制把“Agent 到底要做什么”从 HTTP 元数据下沉到线级语义SQL 侧覆盖SELECT到DROP的完整动词谱系与表/函数维度K8s 侧覆盖具名/集合、命名空间、子资源与动词映射。配合 MCP 透明代理的自动填充与default_registry的可扩展注册治理团队可以用一套 YAML 规则在 Python、Rust、TypeScript、.NET、Go 各 SDK 间保持一致的策略表达将破坏性 SQL、生产命名空间写入、exec子资源等高危操作在规则求值层直接阻断。【免费下载链接】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），仅供参考 
长河君 · 2026/9/18 2:27:13 · 阅读 1.2k Agent Governance Toolkit 协议面（Protocol Facets）：面向线级语义的 SQL 与 Kubernetes 策略评估指南
Agent Governance Toolkit 协议面Protocol Facets面向线级语义的 SQL 与 Kubernetes 策略评估指南【免费下载链接】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 Agent 在执行数据库查询、Kubernetes API 调用等高危操作时传统策略引擎只能看到工具名或 HTTP 元数据无法识别一条 SQL 到底是SELECT还是DROP、一个 K8s 请求是否指向production命名空间。Agent Governance Toolkit 的Protocol Facets协议面机制在PolicyEngine.evaluate()进入规则匹配之前从原始协议上下文SQL 语句、Kubernetes API 路径中结构化提取sql.*与k8s.*字段并合并进评估上下文使 YAML 规则可以用点号dot-notation条件直接引用线级语义。阅读本文后你将掌握 SQL/K8s 面字段的提取规则、HTTP 方法到 Kubernetes 动词的映射、无侵入的 MCP 透明代理集成方式、自定义协议解析器的注册方法以及该模型在 Python、Rust、TypeScript、.NET 等多语言 SDK 中的一致性实现。工作机制规则求值前的结构化提取PolicyEngine.evaluate()的入口逻辑位于 agent-governance-python/agent-mesh/src/agentmesh/governance/policy.py# Populate sql.* and k8s.* fields before rules run from agentmesh.governance.protocol_facets import extract_protocol_facets extract_protocol_facets(context)也就是说每次调用evaluate()时系统都会先运行extract_protocol_facets(context)再评估规则。只要上下文context中包含sql或k8s子字典对应的解析器就会填充结构化字段随后 YAML 规则即可用{field, operator, value}条件引用这些字段。面提取发生在规则匹配之前因此规则作者完全不需要关心原始 SQL 解析或 K8s 路径正则的细节——只需按字段名书写条件。从源码结构看面提取模块 protocol_facets.py 由三部分组成FacetRegistry类按上下文键名持有面提取器提取器收到对应键的子字典并返回要合并的字段提取器内部异常会被捕获并记录日志单个解析器故障不会阻塞策略评估SQL 解析器_extract_sql_facets与 K8s 解析器_extract_k8s_facets模块级default_registry预注册了sql与k8s两个提取器并对外暴露extract_protocol_facets(context, registryNone)便捷函数未传 registry 时使用default_registry。SQL 面sql.*字段定义调用方在上下文中放入context[sql][query]SQL 语句字符串解析器提取以下字段供策略条件引用字段示例值说明sql.verbSELECT、DROP、DELETE大写形式的 SQL 动词sql.targetusers操作的主要表/对象语句涉及的第一张表sql.tablesorders,users引用的全部表的逗号连接列表sql.functionsCOUNT,NOW使用的 SQL 函数名的逗号连接列表依赖说明SQL 解析依赖sqlglotpip install sqlglot。未安装或解析失败时sql.verb会被置为UNKNOWN——这是一种fail-closed失败即拒绝倾向行为避免因解析器缺失而放过危险语句。底层解析原理从 protocol_facets.py 的源码可以看到_extract_sql_facets的实现思路通过sqlglot.parse(query)将语句解析为 AST再依据 AST 节点类型exp.Select、exp.Insert、exp.Update、exp.Delete、exp.Drop、exp.Create、exp.AlterTable、exp.Grant、exp.Merge等判定动词对于无法归类的Command节点通过_SQL_VERB_MAP按命令名映射覆盖TRUNCATE、REVOKE、CALL、EXECUTE、EXPLAIN、WITH等tables通过stmt.find_all(exp.Table)收集functions通过stmt.find_all(exp.Func)收集并转为大写target取语句涉及的第一张表例如SELECT * FROM orders JOIN users ...的 target 是orders。空查询或纯空白查询返回空字符串字段缺少sqlglot时返回UNKNOWN。这些边界行为在 test_protocol_facets.py 中有完整的测试覆盖包括test_no_sqlglot_returns_unknownmock 掉sqlglot导入后断言 verb 为UNKNOWN。示例规则每条规则通过{field, operator, value}匹配单个提取字段。复合检查例如动词 目标组合应拆分为多条带合适优先级的规则rules: - name: deny-destructive-sql condition: {field: sql.verb, operator: in, value: [DROP, TRUNCATE, DELETE]} action: deny priority: 100 - name: deny-schema-changes condition: {field: sql.verb, operator: in, value: [ALTER, GRANT, REVOKE]} action: deny priority: 100 - name: allow-read-only-sql condition: {field: sql.verb, operator: eq, value: SELECT} action: allow priority: 5示例评估上下文engine.evaluate( agent_diddid:example:agent1, context{sql: {query: DROP TABLE production}}, )上面的上下文经extract_protocol_facets处理后sql.verb为DROP、sql.target为production从而命中deny-destructive-sql规则。集成测试 TestPolicyEngineIntegration.test_sql_drop_denied_by_rule 验证了DROP TABLE users会被拒绝且matched_rule deny-drop同文件还验证了DROP TABLE staging非 protected 目标在带default_action: allow的策略下可以放行说明规则可以精确到“目标表”粒度。Kubernetes 面k8s.*字段定义调用方在上下文中放入context[k8s][method]HTTP 方法与context[k8s][path]API Server 路径解析器提取以下字段字段示例值说明k8s.verbget、list、delete、createKubernetes API 动词k8s.resourcepods、deployments资源类型k8s.namespaceproduction命名空间集群级资源为空k8s.namemypod资源名称集合请求为空k8s.subresourceexec、log子资源无则为空HTTP 方法到 Kubernetes 动词的映射HTTP具名资源集合GETgetlistDELETEdeletedeletecollectionPOSTcreatecreatePUTupdateupdatePATCHpatchpatch源码中还额外处理了HEAD具名资源映射为get集合映射为list。从 protocol_facets.py 可以看到判定“具名”还是“集合”的关键在于路径是否包含资源名称_K8S_PATH_PATTERNS中按“最具体优先”排列了 10 条正则覆盖/api/version/...与/apis/group/version/...两类前缀下的具名/集合/命名空间/子资源组合若路径匹配到name组则用_METHOD_TO_VERB_NAMED映射否则用_METHOD_TO_VERB_COLLECTION映射。例如GET /api/v1/namespaces/default/pods/mypod→ verb 为get而GET /api/v1/namespaces/default/pods→ verb 为list见 test_get_named_resource 与 test_get_collection。示例规则rules: - name: deny-k8s-production-namespace condition: {field: k8s.namespace, operator: eq, value: production} action: deny priority: 110 - name: deny-k8s-exec condition: {field: k8s.subresource, operator: eq, value: exec} action: deny priority: 100 - name: deny-k8s-deletecollection condition: {field: k8s.verb, operator: eq, value: deletecollection} action: deny priority: 100 - name: allow-k8s-readonly condition: {field: k8s.verb, operator: in, value: [get, list, watch]} action: allow priority: 5示例评估上下文engine.evaluate( agent_diddid:example:agent1, context{ k8s: { method: DELETE, path: /api/v1/namespaces/production/pods/mypod, } }, )该上下文解析后k8s.verb delete、k8s.namespace production、k8s.resource pods、k8s.name mypod命中deny-k8s-production-namespace。对应集成测试 test_k8s_delete_prod_denied 验证了 production 下的删除被拒绝而 test_k8s_delete_staging_allowed 验证了 staging 下的同形态删除在default_action: allow下放行——这是“按命名空间隔离风险”的典型用法。透明代理集成零应用改造MCP 代理agentmesh proxy会自动从工具调用参数中填充线级协议上下文无需修改任何应用代码名为query或sql的工具参数 → 填充context[sql][query]名为method/http_method配合path/api_path且路径以/api/或/apis/开头 → 填充context[k8s]因此只要在策略文件中定义 SQL 或 K8s 规则这些规则就会自动作用于所有经过代理的工具调用。这对于治理 Agent 通过 MCP 工具触发的数据库操作和集群管理操作非常实用治理规则与业务代码解耦新增协议面规则不需要任何应用侧改动。扩展自定义协议解析器Python注册到 default_registry面提取模块的文档字符串与测试用例test_register_and_extract_custom_protocol均演示了自定义协议的注册方式from agentmesh.governance.protocol_facets import default_registry def extract_redis_facets(redis_ctx: dict) - dict: cmd (redis_ctx.get(command) or ).upper() return {verb: cmd, key: redis_ctx.get(key, )} default_registry.register(redis, extract_redis_facets)之后在评估上下文中传入{redis: {command: FLUSHALL}}即可编写如下规则- name: deny-redis-flush condition: {field: redis.verb, operator: in, value: [FLUSHALL, FLUSHDB]} action: deny值得注意的几个设计细节均有测试佐证提取器接收的是子字典redis_ctx返回的字段会update回该子字典见 FacetRegistry.extract若上下文键的值不是字典例如sql: not-a-dict提取器会被跳过test_non_dict_context_key_skipped提取器抛出的异常会被捕获并记录日志不会阻断策略评估test_extractor_exception_is_swallowed多个提取器按注册顺序依次执行test_multiple_extractors_all_run_in_order也可以构造独立FacetRegistry并通过extract_protocol_facets(ctx, registry...)传入实现不同调用方使用不同解析集合test_custom_registry_passed_to_extract_protocol_facets。完整的规则示例参见 examples/policy-templates/wire-protocol-rules.yamlPython SDK 的agent_control_specification_version: 0.4.0-alpha.1格式策略模板。多语言一致性同一套面模型跨 SDK 可用同一面模型在多个语言 SDK 中保持一致每个 SDK 都暴露FacetRegistry、默认注册表以及extract_protocol_facets等价辅助函数内置sql.*与k8s.*提取器且字段名一致并在策略求值内部自动运行提取器。语言模块 / 包状态Pythonagentmesh.governance.protocol_facets已发布Rustagentmesh::protocol_facets已发布TypeScriptmicrosoft/agent-governance-sdk→protocol-facets仓库内已实现.NETagent-governance-dotnet→AgentGovernance.Policy.ProtocolFacets已实现Goagent-governance-golang已实现说明原文档以 GitHub issue 编号#2553、#2587 等标注各语言状态本仓库中 TypeScript 的实现在 agent-governance-typescript/src/protocol-facets.ts 及其测试 tests/protocol-facets.test.ts并通过 src/index.ts 对外导出Go 与 .NET 的规则示例文件分别位于 agent-governance-golang/examples/wire-protocol-rules.yaml 与 agent-governance-dotnet/examples/Quickstart/wire-protocol-rules.yaml。Rust 用法Rust SDK 在agentmesh::protocol_facets模块下提供同样的面模型FacetRegistry、default_registry()、extract_protocol_facets、extract_sql_facets、extract_k8s_facets并暴露相同的sql.*与k8s.*字段。PolicyEngine::evaluate会在调用方上下文的防御性副本上运行注册表再匹配规则因此不会修改调用方数据。use agentmesh::{PolicyEngine, default_registry}; use serde_yaml::Value; use std::collections::HashMap; let engine PolicyEngine::new(); engine.load_from_yaml(r# version: 1 agent: did:example:agent1 policies: - name: deny-destructive-sql type: capability denied_actions: [*] conditions: sql.verb: [DROP, TRUNCATE, DELETE] #).unwrap(); let mut sub serde_yaml::Mapping::new(); sub.insert(Value::String(query.into()), Value::String(DROP TABLE production.into())); let mut ctx HashMap::new(); ctx.insert(sql.to_string(), Value::Mapping(sub)); let decision engine.evaluate(db.exec, Some(ctx)); // decision PolicyDecision::Deny(...) // Register a custom protocol extractor: default_registry().register(redis, |sub| { let mut m std::collections::HashMap::new(); if let Some(cmd) sub.get(Value::String(command.into())).and_then(|v| v.as_str()) { m.insert(verb.to_string(), Value::String(cmd.to_uppercase())); } m });Rust SDK 的规则条件沿用现有 YAML 映射形态key: value或key: [v1, v2]表示in式成员判断字段名与决策结果与 Python 实现完全一致。完整的 Rust 规则示例见 agent-governance-rust/agentmesh/examples/wire-protocol-rules.yaml。SQL 解析器说明Rust 提取器使用内置的正则分词器覆盖策略规则常用的动词/目标/函数场景。对于复杂的方言级 SQL请通过default_registry().register(sql, ...)注册自定义提取器。.NET 用法.NET SDK 在AgentGovernance.Policy命名空间下暴露同样的面模型FacetRegistry、ProtocolFacets.DefaultRegistry、ProtocolFacets.ExtractProtocolFacets、ProtocolFacets.ExtractSqlFacets、ProtocolFacets.ExtractK8sFacets。PolicyEngine.Evaluate在内部上下文副本上运行注册表调用方只需填充原始sql/k8s子字典——调用方自己的字典不会被修改。using AgentGovernance.Policy; var engine new PolicyEngine(); engine.LoadYaml( apiVersion: governance.toolkit/v1 name: sql-guard scope: global default_action: allow rules: - name: deny-destructive-sql condition: sql.verb DROP action: deny priority: 100 ); var decision engine.Evaluate(did:mesh:agent1, new Dictionarystring, object { [sql] new Dictionarystring, object { [query] DROP TABLE production }, }); // decision.Allowed false, decision.MatchedRule deny-destructive-sql // Register a custom protocol extractor: ProtocolFacets.DefaultRegistry.Register(redis, sub { var cmd sub.TryGetValue(command, out var v) ? v?.ToString() ?? : ; return new Dictionarystring, object { [verb] cmd.ToUpperInvariant() }; });.NET 的规则条件使用表达式字符串格式sql.verb DROP、点路径字段引用、and/or组合字段名与高层决策结果与 Python 实现一致存在少量规则语法差异——尤其是 .NET 的in运算符引用的是列表值上下文字段而非 YAML 字面量列表因此多动词规则建议拆分为单独的检查参见 agent-governance-dotnet/examples/Quickstart/wire-protocol-rules.yaml 中的拆分写法如deny-destructive-sql-drop、deny-destructive-sql-truncate、deny-destructive-sql-delete各自独立成条。SQL 解析器说明.NET 提取器同样使用内置正则分词器并非完整的 SQL 解析器无法覆盖所有方言构造在高保障环境中建议通过ProtocolFacets.DefaultRegistry.Register(sql, ...)注册基于真正 SQL 解析器的自定义提取器。实战建议与注意事项fail-closed 语义SQL 解析在缺少sqlglot或解析失败时返回UNKNOWN动词。治理生产环境的 Agent 时应把UNKNOWN纳入 deny 规则或依赖default_action: deny避免“解析不出就放行”的漏洞。复合条件用多条规则表达Python 的{field, operator, value}每条规则匹配单个字段动词 目标的复合检查应拆分为多条规则并用优先级排序Rust 用conditions映射、.NET 用表达式字符串语法形态各有差异跨语言迁移时留意 各语言规则示例 的差异。面提取不会污染调用方数据Python 端extract_protocol_facets就地更新上下文并返回同一字典对象test_returns_same_dictRust 与 .NET 则在内部副本上运行——调用方无需担心原始上下文被修改或重复提取产生副作用。代理层自动生效通过agentmesh proxy透传的 MCP 工具调用只要参数命名符合query/sql与method/path/api/、/apis/前缀约定规则即可零改造生效自定义协议可遵循同样的参数命名约定并注册对应提取器。总结Wire-Protocol-Aware Policy Evaluation 通过协议面机制把“Agent 到底要做什么”从 HTTP 元数据下沉到线级语义SQL 侧覆盖SELECT到DROP的完整动词谱系与表/函数维度K8s 侧覆盖具名/集合、命名空间、子资源与动词映射。配合 MCP 透明代理的自动填充与default_registry的可扩展注册治理团队可以用一套 YAML 规则在 Python、Rust、TypeScript、.NET、Go 各 SDK 间保持一致的策略表达将破坏性 SQL、生产命名空间写入、exec子资源等高危操作在规则求值层直接阻断。【免费下载链接】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），仅供参考