Agent Governance Toolkit 实战用 Azure Functions 作为 Foundry AI Gateway 的策略决策点PDP【免费下载链接】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本文基于 Agent Governance Toolkit 仓库中的可运行参考示例 examples/foundry-ai-gateway-pdp/README.md讲解如何用 Azure API ManagementAI Gateway作为策略执行点PEP、Azure Functions 作为策略决策点PDP把 Microsoft Foundry prompt-based agent 的模型流量与 MCP 工具流量统一收敛进同一个治理边界。读完本文你将掌握一套可直接落地的 Azure 原生 PEP/PDP 模式版本化决策契约、默认 fail-closed 的失败语义、APIM 策略片段、Python Function 决策实现、Bicep 一键部署以及用于验证自身 SLO 的延迟压测工具。背景Foundry 模型与 MCP 工具流量不对称的治理缺口Microsoft Foundry prompt-based agent 在服务端调用 MCP 工具时工具调用发生在 Foundry 后端天然不会像模型请求那样经过 AI GatewayAzure API Management的策略边界。这就形成了一种不对称的治理姿态模型请求经过集中式策略执行而 MCP 工具调用可能绕过它。仓库中的架构决策记录 ADR-0026 明确指出这一缺口对应 RFC #2470 提出的问题并给出了 AGT 推荐的解决方案使用 Azure 原生组件AI Gateway Azure Functions作为统一的 PEP/PDP 边界让用户在等待 Foundry 原生支持之前就能落地治理。该模式需要满足五条硬性约束必须能与现有 APIM 策略原语send-request、choose、set-header组合不依赖新网关特性决策延迟会直接叠加到每次模型/工具调用上因此 PDP 契约必须支持短路和缓存PDP 处于运行时关键路径必须默认 fail-closedfail-open 只能显式选择发送给 PDP 的请求上下文必须是最小必要集默认不发送原始 prompt以限制爆炸半径契约从第一天起就要版本化以便集成演进时不会破坏已部署的网关策略。架构总览PEP/PDP 分离与 MCP 流量收敛本示例的结构非常精简一个 APIM 策略片段、一个 Python Function、一个 Bicep 模板、一个微小的延迟压测工具。Foundry prompt-based agent │ │ model.invoke / tool.invoke ▼ ┌──────────────────────────┐ ┌──────────────────────────┐ │ AI Gateway (APIM) │ POST │ Azure Function (PDP) │ │ - PEP │ ───────▶│ /api/decide │ │ - send-request policy │ ◀───────│ decision contract v1.0 │ │ - fail-closed default │ └──────────────────────────┘ └─────────────┬────────────┘ │ allow / deny / allow_with_conditions / require_approval ▼ Foundry / MCP tool / model关键设计MCP 流量如何留在网关边界内。把 APIM 网关 URL 注册为面向 Foundry 暴露的 MCP server endpoint这样后端工具调用就会与模型流量走同一条 PEP。ADR-0026 记录了这种做法的依据同时把 sidecar MCP proxy 列为不推荐的回退方案。完整目录布局如下examples/foundry-ai-gateway-pdpfoundry-ai-gateway-pdp/ ├── README.md # 示例说明本文主体 ├── azure.yaml # azd 配置 ├── infra/ │ └── main.bicep # APIM Function App managed identity ├── policy/ │ └── pdp-callout.xml # APIM 策略片段send-request 到 PDP ├── function/ │ ├── host.json │ ├── requirements.txt │ └── decide/ │ ├── __init__.py # HTTP 触发的 PDP │ └── function.json └── load/ └── harness.py # 延迟 / SLO 冒烟测试决策契约 v1.0请求与响应格式契约是这套模式的承重结构。请求APIM → Function格式{ schemaVersion: 1.0, agentId: agent-7f3c, callerIdentity: usercontoso.com, tenantId: 00000000-0000-0000-0000-000000000000, environment: prod, operation: tool.invoke, target: github.create_issue, inputDigest: sha256:9af1..., correlationId: c-abc123, traceparent: 00-... }字段说明字段含义备注schemaVersion契约版本当前固定1.0agentIdAgent 标识从x-agent-id请求头透传callerIdentity调用者身份来自 APIM 的context.User?.EmailtenantId租户 ID来自订阅上下文environment环境标签来自命名值pdp-environmentdev/test/prodoperation操作类型model.invoke或tool.invoketarget目标模型名或 MCP 工具名如github.create_issueinputDigest输入摘要prompt 或工具参数的SHA-256默认不发送原始文本correlationId关联 ID贯穿全链路审计traceparentW3C 追踪上下文便于端到端重建响应Function → APIM格式{ decision: allow_with_conditions, reasonCode: tool.allowlisted.with_approval, conditions: [require_human_review], auditAnnotations: {risk: medium}, ttlSeconds: 30 }语义要点decision枚举allow|deny|allow_with_conditions|require_approvalreasonCode机器可读的决策原因conditions[]可选条件列表如require_human_reviewauditAnnotations可选的审计注解ttlSeconds网关缓存提示网关可以按(agentId, operation, target, inputDigest)为键缓存相同决策。inputDigest的设计意图默认只把 prompt/工具输入的摘要送过 PEP/PDP 边界原始文本不出网关需要完整文本的 PDP 必须显式选择并承担相应的数据处理义务。这是控制数据爆炸半径的核心机制。失败语义默认 fail-closedfail-open 需显式选择网关在以下情况必须失败关闭fail-closed到 Function 的传输错误或超时非 2xx 响应响应 schema 不匹配decision缺失或为未知值任何非allow*的决策。fail-open仅支持作为按路由的显式选择命名值pdp-fail-opentrue且必须限定在非敏感操作上。本示例的默认策略是 fail-closed——这是治理边界正确默认值用可用性换取执行完整性。APIM 策略片段逐段拆解policy/pdp-callout.xml 是可直接粘贴的 drop-in 片段放入面对 Foundry 流量的 API 或 operation 策略的inbound段即可。它分为四个步骤步骤 1构造最小决策请求信封v1.0set-variable namepdpCorrelationId value(context.RequestId) / set-variable namepdpInputDigest value{ var body context.Request.Body?.Asstring(preserveContent: true) ?? string.Empty; using (var sha System.Security.Cryptography.SHA256.Create()) { var hash sha.ComputeHash(System.Text.Encoding.UTF8.GetBytes(body)); return sha256: BitConverter.ToString(hash).Replace(-, string.Empty).ToLowerInvariant(); } } /关联 ID 直接取context.RequestId输入摘要用preserveContent: true读取请求体并就地计算 SHA-256既保证后续后端还能读到原始 body又不把原文透传给 PDP。步骤 2调用 PDP2 秒超时失败即关闭send-request modenew response-variable-namepdpResponse timeout2 ignore-errortrue set-url{{pdp-base-url}}/api/decide/set-url set-methodPOST/set-method authentication-managed-identity resource{{pdp-aad-audience}} / set-header nameContent-Type exists-actionoverride valueapplication/json/value /set-header set-header nametraceparent exists-actionoverride value(context.Request.Headers.GetValueOrDefault(traceparent, string.Empty))/value /set-header set-body{ return new JObject( new JProperty(schemaVersion, 1.0), new JProperty(agentId, context.Request.Headers.GetValueOrDefault(x-agent-id, string.Empty)), new JProperty(callerIdentity, context.User?.Email ?? string.Empty), new JProperty(tenantId, context.Subscription?.Id ?? string.Empty), new JProperty(environment, {{pdp-environment}}), new JProperty(operation, context.Request.Headers.GetValueOrDefault(x-agt-operation, model.invoke)), new JProperty(target, context.Request.Headers.GetValueOrDefault(x-agt-target, string.Empty)), new JProperty(inputDigest, (string)context.Variables[pdpInputDigest]), new JProperty(correlationId, (string)context.Variables[pdpCorrelationId]), new JProperty(traceparent, context.Request.Headers.GetValueOrDefault(traceparent, string.Empty)) ).ToString(); }/set-body /send-request几个值得注意的实现事实timeout2秒刻意收紧PDP 必须快否则就 fail-closed认证使用managed identity面向pdp-aad-audience取 token不使用 Function key请求头x-agent-id、x-agt-operation、x-agt-target分别映射到agentId、operation、target其中operation缺省为model.invoketraceparent从入站请求透传给 PDP实现端到端追踪。步骤 3传输/超时错误的 fail-closed 强制choose when condition(((IResponse)context.Variables[pdpResponse]) null || ((IResponse)context.Variables[pdpResponse]).StatusCode 300) choose when condition({{pdp-fail-open}} true) set-header namex-agt-pdp-failmode exists-actionoverride valuefail-open/value /set-header /when otherwise return-response set-status code503 reasonPDP unavailable / ... /return-response /otherwise /choose /when /choose响应为 null超时/传输错误或状态码 ≥ 300 时默认直接返回503 PDP unavailable并附带x-agt-correlation-id便于审计仅当命名值pdp-fail-open为true时才放行并在响应头标注x-agt-pdp-failmode: fail-open。步骤 4解析决策并执行set-variable namepdpDecision value{ var resp (IResponse)context.Variables[pdpResponse]; if (resp null) { return deny; } try { var body resp.Body.AsJObject(preserveContent: true); return (string)body[decision] ?? deny; } catch { return deny; } } /解析失败一律回落为deny白名单之外的任何值deny、require_approval、未知值、畸形响应全部拒绝。随后的choose分支将allow/allow_with_conditions→ 放行到后端require_approval→ 返回202 Approval required其余含deny→ 返回403 Denied by PDP。两种拒绝路径都会带上x-agt-correlation-id响应头和 JSON 错误体含correlationId保证审计可重建。Python Function PDP 实现function/decide/init.py 实现契约 v1.0。它的决策逻辑刻意保持简单——真正承重的是契约本身请求/响应形态、fail-closed 语义、ttl 提示授权/合规/风险逻辑由你在生产环境替换。核心校验链如下SCHEMA_VERSION 1.0 DENIED_TOOLS {github.delete_repo, azure.delete_subscription} APPROVAL_REQUIRED_TOOLS {github.create_issue, github.merge_pr}入口main(req)依次执行JSON 解析失败 →400 invalid_jsonschemaVersion不匹配 →400 unsupported_schema_version必填字段校验agentId、callerIdentity、operation、target、inputDigest缺失则返回400 missing_fields:...inputDigest形状校验必须以sha256:开头且总长为sha256: 64 位十六进制operation必须为model.invoke或tool.invoke否则400 unsupported_operation。决策分支示例逻辑生产环境应从 Key Vault、App Configuration、OPA 等策略存储读取而非硬编码tool.invoke且 target 在DENIED_TOOLS→denyreasonCode: tool.denylistedttlSeconds: 300tool.invoke且 target 在APPROVAL_REQUIRED_TOOLS→require_approvalconditions: [require_human_review]ttlSeconds: 0其余 →allowreasonCode: default.allowttlSeconds: 30。每次决策都会以结构化日志输出pdp_decision correlationId... decision... reasonCode... operation... target...配合 host.json 中启用的 Application Insights 采样排除Request类型形成审计痕迹。函数的绑定在 function/decide/function.json 中定义HTTP POST 触发authLevel: function路由为decide运行时依赖仅azure-functions1.21.3见 requirements.txt。基础设施部署Bicep 模板与 azd 命令infra/main.bicep 是一个参考部署模板用于评估而非生产一次预置Storage AccountFunction 运行时Application Insights Log AnalyticsPerGB2018保留 30 天Linux Consumption Function App系统托管标识Python 3.11minTlsVersion: 1.2、ftpsState: Disabled、仅 HTTPSAPI ManagementDeveloper SKU系统托管标识四个 APIM 命名值pdp-base-url指向 Function 默认主机名、pdp-aad-audienceapi://pdpAadAppId、pdp-environment、pdp-fail-open默认false。纵深防御的关键Easy AuthauthsettingsV2要求每个调用都携带 Entra tokenrequireAuthentication: true、未认证返回 401、allowedAudiences: [api://pdpAadAppId]。结合 Function 级authLevel即使宿主 key 泄露请求仍需要一个 audience 匹配的合法 Entra token。生产部署还应改用 Premium / Elastic Premium 计划以获得可预测的 PDP 延迟、启用 VNet 集成与私有端点。azure.yaml 声明了pdp服务Python FunctionBicep 基础设施位于infra/目录。运行步骤如下cd examples/foundry-ai-gateway-pdp azd env new agt-pdp-dev azd env set PDP_AAD_APP_ID your-entra-app-client-id azd up # 预置 APIM Function App Easy Auth azd deploy pdp # 部署 PDP 代码 # 直接冒烟测试 PDP需要 host key Entra token python load/harness.py --url https://fn-host/api/decide --rps 20 --duration 30前置条件azd、funcCore Tools、Python 3.11、Azure 订阅以及一个用于 Easy Auth 前置 PDP 的Entra ID 应用注册记录其 Application/Client ID作为pdpAadAppId传入。关于 APIM API 接线的重要说明Bicep 只预置 APIM 与命名值把 Foundry/模型/MCP API 导入 APIM 并挂接策略片段是刻意保留的手动步骤——生产部署通常已拥有自己的 API 定义示例不想与之冲突。片段是 drop-in 的粘贴到前端 Foundry 流量的 API 或 operation 策略的inbound段即可。用延迟压测工具验证自己的 SLOload/harness.py 是一个极简延迟工具向 PDP或经 APIM如果--url指向网关发送合成决策请求报告 p50/p95/p99 延迟与决策分布用于在采纳该模式前验证自身 SLO。python load/harness.py --url https://fn-host/api/decide --rps 20 --duration 30参数--url必填PDP decide 端点、--rps默认 10每秒请求数、--duration默认 15 秒运行时长。实现要点可复现样本目标混合了model.invoke/gpt-4o、tool.invoke/github.create_issue、tool.invoke/github.delete_repo、tool.invoke/search.web覆盖 allow / require_approval / deny / allow 四种路径用secrets.token_hex(32)生成 32 字节十六进制合成摘要——形状与真实 SHA-256 一致PDP 只校验形状不校验哈希基于asynciorun_in_executor实现并发压测按 nearest-rank 计算百分位非 2xx 响应归入http_code超时/网络错误归入transport_error便于直接观察失败路径。示例的期望基线在 Premium 计划且实例已预热的情况下PDP 一跳带来的 p50 延迟低于约 25 msdeny与require_approval决策在网关分别以403和202呈现。安全态势总结APIM 使用**系统托管标识 Easy AuthMicrosoft Entra ID**向 Function 认证不使用 Function keysBicep 模板中已禁用默认情况下只有 prompt/工具输入的摘要跨过 PEP/PDP 边界所有决策都会产出带有correlationId和traceparent的审计记录可端到端重建。不在范围内跟踪中的后续事项一等公民的integrations/foundry-ai-gateway/组件与类型化 PDP SDK——待契约经设计伙伴验证并与 Foundry 产品团队对齐后推进APIM 无法直接前置 MCP 场景下的 sidecar MCP proxy 变体带区域缓存复制的多区域 active/active PDP。快速索引内容路径示例总说明examples/foundry-ai-gateway-pdp/README.md架构决策记录docs/adr/0026-foundry-ai-gateway-functions-pdp.mdAPIM 策略片段examples/foundry-ai-gateway-pdp/policy/pdp-callout.xmlPython Function PDPexamples/foundry-ai-gateway-pdp/function/decide/init.pyBicep 部署模板examples/foundry-ai-gateway-pdp/infra/main.bicep延迟压测工具examples/foundry-ai-gateway-pdp/load/harness.py【免费下载链接】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），仅供参考