代理钩子

代理挂钩是一流的代理框架功能，可用于在代理的执行中明确定义的点应用治理和运行时控制。它实现与框架无关的 AGENT-HOOKS-0.1 协定，因此策略引擎、审批网关、预算防护、内容筛选器和出口控件可以面向一个常见的控制面。

Important

代理挂钩是控制平面，而不是遥测平面。每个拦截器返回判决。在 enforce 模式中，框架对该判决执行操作;在 evaluate_only 模式下，它会记录判决而不更改执行。对被动跟踪、指标和日志使用可观测性。

代理挂钩尚不适用于.NET。使用代理中间件、工具审批和代理安全性将运行时控制添加到.NET代理。

代理挂钩在 Python 中是实验性的。工厂会在首次使用时发出 ExperimentalWarning ，其 API 可以在正式发布之前进行更改。

何时使用代理挂钩

独立开发的控件需要跨代理输入、模型调用、工具调用和最终输出共享的可强制协定时使用代理挂钩。

能力	将其用于
代理挂钩	在代理生命周期内，标准化策略决策、转换、审批、预算和出口控制。
代理中间件	不需要代理挂钩协定或其核心运行时保证的应用程序特定的交叉行为。
借助 FIDES 实现代理安全	不受信任的或机密内容的确定性信息流标签和策略。
工具审批	人工确认单个函数工具调用。
可观察性	不控制执行的被动跟踪、指标和日志。

代理框架强制实施的内容

将代理挂钩添加到代理时，代理框架会在代理运行、模型调用和工具调用之间应用协调的强制边界。运行时提供以下保证：

失败关闭： 拒绝会阻止受保护的操作。无效上下文、无效判决、拦截器失败和强制失败不会无提示绕过控制。
转换写回： 转换会更改执行实际使用的本机消息、工具参数、工具结果或最终响应。如果无法应用转换，运行将失败。
缓冲流： 在完整的模型响应和最终输出通过截获点之前，没有响应更新到达调用方。
判决封闭持久性： 持久性等待涵盖它的判决。标准运行后持久性等待 output;每个服务调用历史记录暂留等待每个 post_model_call。
完成捆绑包安装： 代理、聊天和函数部件作为一个单元进行安装，因此无法意外配置不完整的强制边界。

协定是合作的，而不是进程隔离边界。侦听器在主机进程中运行，并接收做出决策所需的内容。仅注册你信任的侦听器。

安装代理挂钩

安装核心包的可选 agent-hooks 附加内容：

pip install "agent-framework-core[agent-hooks]"

如果使用 uv：

uv add "agent-framework-core[agent-hooks]"

依赖项 agent-hooks-sdk 是延迟导入的。 agent_framework除非创建代理挂钩中间件捆绑包，否则导入不会加载 SDK。

注释

特 agent-hooks 意不包括在 agent-framework-core[all]内。若要启用此实验性控制图面，请显式安装它。

添加侦听器

侦听器接收（ agent_hooks.AgentContext 规范的上下文映射，而不是 agent_framework.AgentContext 代理中间件使用的上下文映射），并返回判决。以下拦截器阻止包含单词 secret的最终输出。该示例假定 client 已配置代理框架聊天客户端。

from agent_framework import Agent, create_agent_hooks_middleware
from agent_hooks import ALLOW, AgentContext, InterceptionBlocked, Verdict


class SecretEgressGuard:
    def intercept(self, context: AgentContext) -> Verdict:
        if (
            context["interception_point"] == "output"
            and "secret" in str(context["target"]).lower()
        ):
            return Verdict.deny(
                reason="secret_in_output",
                message="The final response contains restricted content.",
            )
        return ALLOW


hooks = create_agent_hooks_middleware(
    {"secret-egress": SecretEgressGuard()},
)

agent = Agent(
    client=client,
    instructions="You are a helpful assistant.",
    middleware=[hooks],
)

try:
    response = await agent.run("Summarize the account details.")
except InterceptionBlocked as exc:
    print(f"Blocked: {exc.result.verdict.reason}")

将捆绑包作为代理列表的 middleware 一个元素传递。在每个代理上安装完全相同的一个代理挂钩捆绑包。

截获点

代理框架自动发出适用的拦截点：

拦截点	发出时	转换目标
`agent_startup`	在代理挂钩会话中的第一个输入之前	不可转换
`input`	当外部请求进入代理时	输入内容和角色
`pre_model_call`	在每个模型请求之前	发送到模型的消息
`post_model_call`	每个完整的模型响应后	响应内容、框架执行的工具调用和完成原因
`pre_tool_call`	在每个框架执行的工具调用之前	工具参数
`post_tool_call`	工具成功或失败后	工具结果
`output`	在最终响应到达调用方之前	最终响应内容
`agent_shutdown`	代理挂钩会话完成、失败或取消时	不可转换

调用工具的运行通常发出：

agent_startup input pre_model_call post_model_call →→→→ pre_tool_call → post_model_callpost_tool_callpre_model_calloutput →→→→agent_shutdown

判决

合同有三个决定： allow， deny和 transform。 Python SDK 还提供警告和可解除拒绝的帮助程序。

Result	Python API	Behavior
允许	`ALLOW` 或 `Verdict(decision=Decision.ALLOW)`	继续执行目标不变。
允许出现警告	`Verdict.warn(...)`	继续并在拦截记录中包含警告。
拒绝	`Verdict.deny(...)`	阻止受保护的操作。
拒绝等待审批	`Verdict.escalate(...)`	除非配置的审批解析程序返回许可证判决，否则阻止。
转换	`Verdict(decision=Decision.TRANSFORM, transform=Transform(...))`	重写下面的 `$target`值，然后继续重写的值。

运行级别和模型级拒绝引发 InterceptionBlocked 并阻止受保护的结果到达调用方或下一阶段。在工具接缝处，策略拒绝会阻止工具操作或放弃其结果，并将包含策略原因（未拒绝的目标有效负载）的控制错误返回模型。这允许代理循环继续。主机或强制失败会停止运行。

应用转换

转换路径必须从 . $target. 例如，侦听器可以替换最终响应内容：

from agent_hooks import ALLOW, AgentContext, Decision, Transform, Verdict


class OutputRedactor:
    def intercept(self, context: AgentContext) -> Verdict:
        if context["interception_point"] != "output":
            return ALLOW

        return Verdict(
            decision=Decision.TRANSFORM,
            reason="redacted_output",
            transform=Transform(
                path="$target.content",
                value="[Response removed by policy]",
            ),
        )

转换应用于 Agent Framework Content 值，保留支持的丰富内容，而不是将每个值减少为纯文本。格式不正确的路径或不兼容的替换失败，而不是继续使用原始值。

工具审批和参数转换

代理框架工具审批和代理挂钩审批接缝是单独的机制。对于具有 approval_mode="always_require" 的函数工具，Agent Framework 会在函数中间件运行之前创建人工审批请求。因此，转换 pre_tool_call 可以在用户批准原始值后更改参数。

Warning

不要将参数 pre_tool_call 转换为使用 approval_mode="always_require"的工具。转换工具调用post_model_call，使框架审批请求包含转换的值，或通过代理挂钩resolver返回Verdict.escalate(...)pre_tool_call并解析审批。

流式处理和持久性

代理挂钩保留流式处理 API，但使用缓冲输出语义。代理框架组装完整的模型响应、发出 post_model_call、组装最终代理响应，并在发布任何更新之前发出 output 。如果任一点都拒绝响应，则调用方不会收到部分更新。

此行为将按令牌的延迟交易为失败的输出强制实施。输出转换也会反映在最终向调用方发布的更新中。

持久性由覆盖持久性操作的拦截点封闭：

默认情况下，历史记录和其他运行后提供程序的工作等待 output 判决。拒绝的输出不会持久化，并且转换后将持久保存输出转换。
在构造函数上Agent设置require_per_service_call_history_persistence=True时，或client.as_agent(...)在其判决允许后post_model_call保留每个模型交换。以后 output 的拒绝不会回滚已允许的历史记录。
对于默认的运行后持久性，重试尝试仍保留在最终 output 决策之后。按服务调用模式改为保留传递 post_model_call的每个模型响应。

Important

如果模型内容不能成为持久内容，请在何时require_per_service_call_history_persistence=True强制实施该策略post_model_call。仅输出出口策略可保护到达调用方的内容，但它不会追溯删除已允许和保留的 post_model_call模型交换。

会话和审核记录

默认情况下，每个代理运行都会创建一个代理挂钩会话。 agent_startup 并 agent_shutdown 括号运行，记录接收一个会话 ID，其顺序单调递增。

用于 record_sink 接收每个 InterceptionRecord：

records = []

hooks = create_agent_hooks_middleware(
    {"secret-egress": SecretEgressGuard()},
    record_sink=records.append,
)

截获记录捕获决策、原因、拦截器摘要、模式、标识和序列，而无需将截获的有效负载复制到审核记录中。侦听器本身仍接收完整上下文。

使用一个会话跨多个运行

当应用程序拥有较长生存期的代理挂钩会话（例如具有一个审批账本的会话）时使用 create_agent_hooks_middleware_from_emitter() ：

from agent_framework import Agent, create_agent_hooks_middleware_from_emitter
from agent_hooks import AgentContextBuilder, InterceptionEmitter


emitter = InterceptionEmitter().register(SecretEgressGuard())
builder = AgentContextBuilder(
    agent_id="support-agent",
    framework="agent-framework",
    session_id="conversation-42",
)

hooks = create_agent_hooks_middleware_from_emitter(emitter, builder)
agent = Agent(client=client, middleware=[hooks])

await emitter.emit(builder.agent_startup(tools_registered=[]))
await agent.run("First turn")
await agent.run("Second turn")
await emitter.emit(builder.agent_shutdown(reason="completed"))

在此窗体中，应用程序配置发出器并拥有启动、关闭和错误清理。中间件从 input 中 output发出每运行点。

配置强制

create_agent_hooks_middleware() 接受以下控件：

参数	Purpose
`interceptors`	侦听器序列或名称到拦截器映射。至少需要一种属性。
`resolver`	通过审批渠道解决可解除的拒绝。如果没有解析程序，拒绝仍然有效。
`mode`	`"enforce"` 应用判决。 `"evaluate_only"` 记录将发生的情况，但允许执行每个操作。
`composition`	选择合并多个拦截器判决的方式。
`identity_provider`	生成内容绑定上下文标识。默认值为 `"jcs-sha256"`。
`timeout`	可等待调用的按侦听器和解析程序超时。默认值为 5 秒。阻止事件循环的同步侦听器或解析程序无法被此超时抢占。
`record_sink`	接收每个无有效负载拦截记录。

默认组合按顺序 first_deny 排列，审批配置为停止折叠。因此，侦听器顺序很重要：在可以请求审批的控件之前，放置必须始终运行的控件。选择另一个组合配置文件之前，请参阅代理挂钩生产清单。

使用仅评估模式推出

用于 evaluate_only 在强制实施之前测量策略行为：

hooks = create_agent_hooks_middleware(
    {"secret-egress": SecretEgressGuard()},
    mode="evaluate_only",
    record_sink=records.append,
)

在此模式下，侦听器运行和记录包括其判决，但不会阻止或转换任何操作。不要将部署描述 evaluate_only 为强制治理。

组合规则

将捆绑包放在代理的中间件列表中，使其形成最外部的强制边界：

agent = Agent(
    client=client,
    middleware=[
        create_agent_hooks_middleware([SecretEgressGuard()]),
        application_middleware,
    ],
)

遵循以下规则：

为每个代理安装一个代理挂钩捆绑包。堆积捆绑包被拒绝。
使捆绑包保持不变。无法单独安装其代理、聊天和函数中间件。
在聊天客户端上或通过上下文提供程序安装捆绑包，而不是直接在聊天客户端上安装捆绑包 Agent。
捆绑包位于强制边界之外之前放置的中间件。将外部位置视为外部信任。
当每个嵌套代理的内部模型和工具活动也需要拦截时，为每个嵌套代理提供自己的捆绑包。

当前限制

仅Python：代理挂钩尚未在 .NET 或 Go SDK 中实现。
实验性 API： 在正式发布之前，工厂签名和行为可能会更改。
缓冲流： 更新不会由令牌发布，因为输出必须在关闭失败的判决之前完成。
托管工具： 模型提供程序执行的工具不会通过 Agent Framework 的函数调用接缝。它们的调用和输出浮出水面post_model_call，但pre_tool_callpost_tool_call无法阻止提供程序的服务器端执行。
协作边界： 代理挂钩不会沙盒拦截器，也不会防范恶意主机。未涵盖绕过受保护的代理管道的代码路径。
侦听器可用性会影响代理可用性： 在强制模式下，拦截器故障或超时会按设计阻止受保护的操作。

有关生产推出、失败原因和警报指南，请参阅代理挂钩操作 Runbook。

代理挂钩尚不适用于 Go。使用代理中间件、工具审批和代理安全将运行时控制添加到 Go 代理。

后续步骤

了解代理管道

反馈

此页面是否有帮助？

Last updated on 2026-08-14