实践｜Hooks，把检查和审计接到执行入口

先给新增行为找到位置

思想篇留下了一个必须写进代码的区别：运行前可以阻止，运行后只能观察已经取得的结果。我们继续读取并运行 main.py，新增维护检查与审计；模型循环、工具注册和第三课权限入口沿用原实现。

正式代码在独立公开仓库 examples/s04-hooks。本次只有三个核心函数：run_hooks 执行回调链，execute_hooked 把链放到原执行入口两侧，audit_event 从观察数据提取记录字段。它们下面展示的正文都按符号从公开源码提取。

事件契约先定下来：before 的 None 表示继续，非空字符串表示阻止；异常和错误返回类型也阻止。after 必须返回 None，异常或错误返回只产生警告，后续观察者继续。字符串 allow 也是非空阻止理由，不能用它覆盖权限。

代码与图共享已有断点游标。浏览器原样运行函数正文，回调和异常真实执行；宿主回答、文件与进程使用可展开的公开替身。切换教学输入会重置回放与断点，需要在同一源码行重新设置。

run_hooks：顺序也是契约的一部分

先选“首个阻止”，在 reason = callback(...) 处设置断点。继续时会再次遇到同一行，但前一个回调是否返回阻止理由，决定了还能不能看到下一次调用。把 return {"blocked": reason, ...} 与循环末尾的返回对照，就能找到短路发生的位置。

tuple(...) 固定本次触发的回调序列，deepcopy(payload) 则给每个观察者独立的数据。前者避免执行中追加注册项改变当前遍历，后者避免某个回调修改下一位将看到的请求。两者都没有隔离回调能访问的其他全局状态。

真实执行轨迹回放；回调与异常真实执行，宿主、文件与进程使用公开教学替身。

Pys04/core/runtime.py · run_hooks

未运行

def run_hooks(hooks, event, payload):
    """before 按顺序遇阻即停；after 故障只记警告，并继续其他观察者。"""
    if event not in ("before", "after"):
        raise ValueError("未知 hook 事件")
    warnings = []
    for index, callback in enumerate(tuple(hooks.get(event, ()))):
        try:
            reason = callback(deepcopy(payload))
            if reason is not None:
                if event != "before" or not isinstance(reason, str) or not reason.strip():
                    raise ValueError("before 只能返回 None 或非空拒绝理由；after 必须返回 None")
                return {"blocked": reason, "warnings": warnings}
        except Exception as error:
            message = type(error).__name__ + ": " + str(error)
            if event == "before":
                return {"blocked": "前置检查异常：" + message, "warnings": warnings}
            warnings.append(message)
    return {"blocked": None, "warnings": warnings}

输出

点击行号设置断点，再点“调试”；也可以直接运行。

代码只读；执行发生在浏览器中，不会发送到博客服务器。

教学准备与调用入口

上方源码保持原样。以下准备输入和调用入口，在单独文件名中执行；断点只回放上方源码。

from copy import deepcopy
scenario='pass'
entries=[]
visited=[]
_executions=[]
_approvals=[]
call={"type":"tool_use","id":"run_1","name":"run_main","input":{}}
def _run():
    _executions.append("run_main")
    return "exit_code=0\n42"
tools={"run_main":{"parameters":{},"run":_run}}
def ask_user(shown,reason):
    _approvals.append(shown)
    return scenario!="refused"
def first(payload):
    visited.append("first")
    if scenario=="block":return "维护期间暂停运行。"
    if scenario in ("pre_error","post_error"):raise OSError("审计不可用")
    if scenario=="mutate":
        payload["call"].update(id="changed",name="unknown")
        payload.get("result",{}).update(content="fake")
def second(payload):
    visited.append("second")
hooks={"before":[first,second],"after":[second]}
event="after" if scenario=="post_error" else "before"
payload={"call":call}

def validate_call(call, tools):
    """本课小型参数契约；供分发与后续权限入口共用。"""
    name, arguments = call.get("name"), call.get("input")
    if not isinstance(name, str) or name not in tools:
        return "未知工具。"
    parameters = tools[name]["parameters"]
    if not isinstance(arguments, dict) or set(arguments) != set(parameters):
        return "参数字段不匹配。"
    if any(type(arguments[key]) is not kind for key, kind in parameters.items()):
        return "参数类型不匹配。"
    return None

def execute_tool(call, tools):
    """校验与执行；每个返回值保留原调用 ID。"""
    result = {"type": "tool_result", "tool_use_id": call["id"]}
    error = validate_call(call, tools)
    if error:
        return {**result, "content": error, "is_error": True}
    try:
        output = tools[call["name"]]["run"](**call["input"])
        return {**result, "content": output, "is_error": False}
    except (ValueError, OSError) as error:
        return {**result, "content": str(error), "is_error": True}

def permission_for(call, tools, blocked=()):
    """只判断，不提问、不执行。没有策略的工具默认拒绝。"""
    error = validate_call(call, tools)
    if error:
        return {"decision": "deny", "reason": error}
    name = call["name"]
    if name in blocked:
        return {"decision": "deny", "reason": "宿主策略禁止这个工具。"}
    if name == "read_file":
        return {"decision": "allow", "reason": "允许读取；目录约束仍由读取函数检查。"}
    if name == "run_main":
        return {"decision": "ask", "reason": "运行 main.py 前需要用户允许。"}
    return {"decision": "deny", "reason": "这个工具尚未配置权限策略。"}

def authorize_call(call, tools, ask_user, blocked=()):
    """只有 ask 才调用宿主；必须收到布尔 True 才允许。"""
    verdict = permission_for(call, tools, blocked)
    if verdict["decision"] != "ask":
        return verdict
    # 宿主看到调用副本；不能通过修改显示对象改变后续执行参数。
    answer = ask_user(deepcopy(call), verdict["reason"])
    if answer is True:
        return {"decision": "allow", "reason": "用户允许本次调用。"}
    return {"decision": "deny", "reason": "本次调用未获用户允许。"}

def execute_guarded(call, tools, ask_user, blocked=()):
    """验证、显示与执行使用同一份调用快照；拒绝结果不冒充执行结果。"""
    snapshot = deepcopy(call)
    verdict = authorize_call(snapshot, tools, ask_user, blocked)
    if verdict["decision"] != "allow":
        return {
            "type": "tool_result", "tool_use_id": snapshot["id"],
            "content": "未执行：" + verdict["reason"], "is_error": True,
        }
    return execute_tool(snapshot, tools)


# 调用入口
outcome=run_hooks(hooks,event,payload)
print("回调顺序："+" → ".join(visited))

固定本次回调顺序，每个回调接收独立副本。
before 首个阻止、异常或错误类型结束前置链。
after 故障进入 warnings，继续后续观察者。

切到“前置异常”，异常会变成阻止理由；“后置异常”则在 warnings.append(message) 停下，随后继续下一个观察者。except Exception 捕获的是普通异常，不吞掉 KeyboardInterrupt、进程终止或无限等待。这里没有超时执行器。

我们还检查返回类型。前置回调误写 return False 不会被当作“没有阻止”，而是协议错误；后置回调返回拒绝理由也不会回到过去撤销执行，只形成警告。契约越小，调用者越容易判断分支，不必猜每个回调的“真假值”是什么意思。

execute_hooked：包住权限入口，保留原结果

这层首先复制调用，再触发 before。被阻止时直接生成带原 ID 的未执行反馈；没有被阻止，才调用第三课的 execute_guarded。因此新增 Hook 没有取代权限规则，也没有复制另一套工具分发器。

在“维护阻止”中给 result = execute_guarded(...) 设置断点，它不会命中。切换“权限拒绝”后重新设置，执行会到达这一行，但内部仍可拒绝，程序不会启动。要区分“进入权限入口”与“已经运行程序”，就看返回的具体内容，而不能只数执行器被调用了几次。

真实执行轨迹回放；回调与异常真实执行，宿主、文件与进程使用公开教学替身。

Pys04/core/runtime.py · execute_hooked

未运行

def execute_hooked(call, tools, ask_user, hooks, blocked=()):
    """包住上一课权限入口；后置观察故障不覆盖实际结果，也不重试工具。"""
    snapshot = deepcopy(call)
    before = run_hooks(hooks, "before", {"call": snapshot})
    if before["blocked"] is not None:
        result = {"type": "tool_result", "tool_use_id": snapshot["id"],
                  "content": "未执行：" + before["blocked"], "is_error": True}
    else:
        result = execute_guarded(snapshot, tools, ask_user, blocked)
    after = run_hooks(hooks, "after", {"call": snapshot, "result": result})
    return {"result": result, "warnings": after["warnings"]}

输出

点击行号设置断点，再点“调试”；也可以直接运行。

代码只读；执行发生在浏览器中，不会发送到博客服务器。

教学准备与调用入口

上方源码保持原样。以下准备输入和调用入口，在单独文件名中执行；断点只回放上方源码。

from copy import deepcopy
scenario='pass'
entries=[]
visited=[]
_executions=[]
_approvals=[]
call={"type":"tool_use","id":"run_1","name":"run_main","input":{}}
def _run():
    _executions.append("run_main")
    return "exit_code=0\n42"
tools={"run_main":{"parameters":{},"run":_run}}
def ask_user(shown,reason):
    _approvals.append(shown)
    return scenario!="refused"
def first(payload):
    visited.append("first")
    if scenario=="block":return "维护期间暂停运行。"
    if scenario in ("pre_error","post_error"):raise OSError("审计不可用")
    if scenario=="mutate":
        payload["call"].update(id="changed",name="unknown")
        payload.get("result",{}).update(content="fake")
def second(payload):
    visited.append("second")
hooks={"before":[first,second],"after":[second]}
event="after" if scenario=="post_error" else "before"
payload={"call":call}

def validate_call(call, tools):
    """本课小型参数契约；供分发与后续权限入口共用。"""
    name, arguments = call.get("name"), call.get("input")
    if not isinstance(name, str) or name not in tools:
        return "未知工具。"
    parameters = tools[name]["parameters"]
    if not isinstance(arguments, dict) or set(arguments) != set(parameters):
        return "参数字段不匹配。"
    if any(type(arguments[key]) is not kind for key, kind in parameters.items()):
        return "参数类型不匹配。"
    return None

def execute_tool(call, tools):
    """校验与执行；每个返回值保留原调用 ID。"""
    result = {"type": "tool_result", "tool_use_id": call["id"]}
    error = validate_call(call, tools)
    if error:
        return {**result, "content": error, "is_error": True}
    try:
        output = tools[call["name"]]["run"](**call["input"])
        return {**result, "content": output, "is_error": False}
    except (ValueError, OSError) as error:
        return {**result, "content": str(error), "is_error": True}

def permission_for(call, tools, blocked=()):
    """只判断，不提问、不执行。没有策略的工具默认拒绝。"""
    error = validate_call(call, tools)
    if error:
        return {"decision": "deny", "reason": error}
    name = call["name"]
    if name in blocked:
        return {"decision": "deny", "reason": "宿主策略禁止这个工具。"}
    if name == "read_file":
        return {"decision": "allow", "reason": "允许读取；目录约束仍由读取函数检查。"}
    if name == "run_main":
        return {"decision": "ask", "reason": "运行 main.py 前需要用户允许。"}
    return {"decision": "deny", "reason": "这个工具尚未配置权限策略。"}

def authorize_call(call, tools, ask_user, blocked=()):
    """只有 ask 才调用宿主；必须收到布尔 True 才允许。"""
    verdict = permission_for(call, tools, blocked)
    if verdict["decision"] != "ask":
        return verdict
    # 宿主看到调用副本；不能通过修改显示对象改变后续执行参数。
    answer = ask_user(deepcopy(call), verdict["reason"])
    if answer is True:
        return {"decision": "allow", "reason": "用户允许本次调用。"}
    return {"decision": "deny", "reason": "本次调用未获用户允许。"}

def execute_guarded(call, tools, ask_user, blocked=()):
    """验证、显示与执行使用同一份调用快照；拒绝结果不冒充执行结果。"""
    snapshot = deepcopy(call)
    verdict = authorize_call(snapshot, tools, ask_user, blocked)
    if verdict["decision"] != "allow":
        return {
            "type": "tool_result", "tool_use_id": snapshot["id"],
            "content": "未执行：" + verdict["reason"], "is_error": True,
        }
    return execute_tool(snapshot, tools)

def run_hooks(hooks, event, payload):
    """before 按顺序遇阻即停；after 故障只记警告，并继续其他观察者。"""
    if event not in ("before", "after"):
        raise ValueError("未知 hook 事件")
    warnings = []
    for index, callback in enumerate(tuple(hooks.get(event, ()))):
        try:
            reason = callback(deepcopy(payload))
            if reason is not None:
                if event != "before" or not isinstance(reason, str) or not reason.strip():
                    raise ValueError("before 只能返回 None 或非空拒绝理由；after 必须返回 None")
                return {"blocked": reason, "warnings": warnings}
        except Exception as error:
            message = type(error).__name__ + ": " + str(error)
            if event == "before":
                return {"blocked": "前置检查异常：" + message, "warnings": warnings}
            warnings.append(message)
    return {"blocked": None, "warnings": warnings}


# 调用入口
outcome=execute_hooked(call,tools,ask_user,hooks)
print("实际执行："+("、".join(_executions) or "无"))

先保存调用副本，触发 before。
未阻止才进入第三章权限入口；拒绝也形成带原ID的结果。
有结果后触发 after；原结果和警告分别返回。

最值得停一下的是 after = run_hooks(...)。选择“审计故障”，此时 result 已经含有 42。执行这一行后，after["warnings"] 增加错误；最终返回值同时保存 result 与警告，调用者没有重试工具的分支。

“修改观察副本”会让前置回调更换自己看到的工具名和调用 ID。继续到权限入口时，真正的 snapshot 仍是 run_main 与 run_1。后置观察数据同样复制。不过宿主装配代码和回调本身仍是可信应用代码，副本不是阻挡恶意 Python 的安全边界。

本课 after 包含维护阻止、权限拒绝和工具返回的错误结果。若权限接口抛出意外异常，execute_guarded 没有正常返回，包装层也不会触发 after；异常继续向上传递。它没有 finally 收尾或崩溃审计保证，不应被当成覆盖所有退出路径的审计系统。

audit_event：只记录当前能证实的字段

前置事件尚无结果，只能登记 event、id 和 tool。后置事件才提取 is_error 和 content。同一个调用 ID 把两条记录连起来，但“看到前置记录”仍不能证明工具已经执行。

选择“登记未执行”，观察 entry 如何加入拒绝内容。审计函数不把所有错误都翻译成“程序失败”，也不从工具名推断结果；它保留入口交来的事实。

真实执行轨迹回放；回调与异常真实执行，宿主、文件与进程使用公开教学替身。

Pys04/core/runtime.py · audit_event

未运行

def audit_event(event, payload, entries):
    """只提取审计字段，既不修改请求，也不决定是否允许。"""
    call = payload["call"]
    entry = {"event": event, "id": call["id"], "tool": call["name"]}
    if event == "after":
        result = payload["result"]
        entry["is_error"] = result["is_error"]
        entry["content"] = result["content"]
    entries.append(entry)

输出

点击行号设置断点，再点“调试”；也可以直接运行。

代码只读；执行发生在浏览器中，不会发送到博客服务器。

教学准备与调用入口

上方源码保持原样。以下准备输入和调用入口，在单独文件名中执行；断点只回放上方源码。

from copy import deepcopy
scenario='before'
entries=[]
visited=[]
_executions=[]
_approvals=[]
call={"type":"tool_use","id":"run_1","name":"run_main","input":{}}
def _run():
    _executions.append("run_main")
    return "exit_code=0\n42"
tools={"run_main":{"parameters":{},"run":_run}}
def ask_user(shown,reason):
    _approvals.append(shown)
    return scenario!="refused"
def first(payload):
    visited.append("first")
    if scenario=="block":return "维护期间暂停运行。"
    if scenario in ("pre_error","post_error"):raise OSError("审计不可用")
    if scenario=="mutate":
        payload["call"].update(id="changed",name="unknown")
        payload.get("result",{}).update(content="fake")
def second(payload):
    visited.append("second")
hooks={"before":[first,second],"after":[second]}
event="after" if scenario=="post_error" else "before"
payload={"call":call}

def validate_call(call, tools):
    """本课小型参数契约；供分发与后续权限入口共用。"""
    name, arguments = call.get("name"), call.get("input")
    if not isinstance(name, str) or name not in tools:
        return "未知工具。"
    parameters = tools[name]["parameters"]
    if not isinstance(arguments, dict) or set(arguments) != set(parameters):
        return "参数字段不匹配。"
    if any(type(arguments[key]) is not kind for key, kind in parameters.items()):
        return "参数类型不匹配。"
    return None

def execute_tool(call, tools):
    """校验与执行；每个返回值保留原调用 ID。"""
    result = {"type": "tool_result", "tool_use_id": call["id"]}
    error = validate_call(call, tools)
    if error:
        return {**result, "content": error, "is_error": True}
    try:
        output = tools[call["name"]]["run"](**call["input"])
        return {**result, "content": output, "is_error": False}
    except (ValueError, OSError) as error:
        return {**result, "content": str(error), "is_error": True}

def permission_for(call, tools, blocked=()):
    """只判断，不提问、不执行。没有策略的工具默认拒绝。"""
    error = validate_call(call, tools)
    if error:
        return {"decision": "deny", "reason": error}
    name = call["name"]
    if name in blocked:
        return {"decision": "deny", "reason": "宿主策略禁止这个工具。"}
    if name == "read_file":
        return {"decision": "allow", "reason": "允许读取；目录约束仍由读取函数检查。"}
    if name == "run_main":
        return {"decision": "ask", "reason": "运行 main.py 前需要用户允许。"}
    return {"decision": "deny", "reason": "这个工具尚未配置权限策略。"}

def authorize_call(call, tools, ask_user, blocked=()):
    """只有 ask 才调用宿主；必须收到布尔 True 才允许。"""
    verdict = permission_for(call, tools, blocked)
    if verdict["decision"] != "ask":
        return verdict
    # 宿主看到调用副本；不能通过修改显示对象改变后续执行参数。
    answer = ask_user(deepcopy(call), verdict["reason"])
    if answer is True:
        return {"decision": "allow", "reason": "用户允许本次调用。"}
    return {"decision": "deny", "reason": "本次调用未获用户允许。"}

def execute_guarded(call, tools, ask_user, blocked=()):
    """验证、显示与执行使用同一份调用快照；拒绝结果不冒充执行结果。"""
    snapshot = deepcopy(call)
    verdict = authorize_call(snapshot, tools, ask_user, blocked)
    if verdict["decision"] != "allow":
        return {
            "type": "tool_result", "tool_use_id": snapshot["id"],
            "content": "未执行：" + verdict["reason"], "is_error": True,
        }
    return execute_tool(snapshot, tools)

def run_hooks(hooks, event, payload):
    """before 按顺序遇阻即停；after 故障只记警告，并继续其他观察者。"""
    if event not in ("before", "after"):
        raise ValueError("未知 hook 事件")
    warnings = []
    for index, callback in enumerate(tuple(hooks.get(event, ()))):
        try:
            reason = callback(deepcopy(payload))
            if reason is not None:
                if event != "before" or not isinstance(reason, str) or not reason.strip():
                    raise ValueError("before 只能返回 None 或非空拒绝理由；after 必须返回 None")
                return {"blocked": reason, "warnings": warnings}
        except Exception as error:
            message = type(error).__name__ + ": " + str(error)
            if event == "before":
                return {"blocked": "前置检查异常：" + message, "warnings": warnings}
            warnings.append(message)
    return {"blocked": None, "warnings": warnings}

event="before" if scenario=="before" else "after"
payload["result"]={"is_error":scenario=="rejected","content":"未执行：用户拒绝" if scenario=="rejected" else "exit_code=0\n42"}


# 调用入口
audit_event(event,payload,entries)
outcome=entries
print("审计条目："+str(len(entries)))

提取请求的event、id和tool。
after额外提取is_error和content，包含未执行反馈。
追加宿主内存列表，不改变工具结果。

entries.append(entry) 只是追加宿主内存列表，进程结束就丢失。这足以观察顺序，却没有可靠落盘、唯一事件键或防重复写入。生产日志还应控制内容大小和敏感字段；本课只记录短教学文件和结果，没有将整份对话复制进审计。

注册发生在公开入口的 build_hooks，可以直接读到顺序：

python

hooks = {
    "before": [audit_before, maintenance_check],
    "after": [audit_after],
}

前置审计在维护检查之前，所以被维护阻止的请求也有请求记录。调换两者次序，会改变哪些回调能够到达；不是等价的代码整理。完整入口中 check 就是这里的维护检查，三个回调闭包明确保留在公开源码里。

接回原循环，别把宿主警告塞进工具协议

execute_hooked 返回的是宿主报告，原循环接收的却必须仍是标准 tool_result。装配层负责拆开两者：

python

def execute(call):
    report = execute_hooked(call, tools, approve, hooks, blocked)
    warnings.extend(report["warnings"])
    return report["result"]

outcome = run_loop(ask, execute, max_calls=4, question=QUESTION)

如果直接把整份 report 交回循环，tool_use_id 和内容就不在协议要求的位置。保持 execute(call) → tool_result 这个已有接口，前三课的保存、关联、再次判断和调用上限就都可以复用；审计故障另由宿主显示。

Python 3.10+ 标准库即可运行下面三条离线路径。固定允许回答只消费一次；不带 --decision 时，默认仍在交互终端询问，非交互输入按拒绝处理。

bash

python3 examples/s04-hooks/user_code/practice.py --decision allow
python3 examples/s04-hooks/user_code/practice.py --decision allow --maintenance
python3 examples/s04-hooks/user_code/practice.py --decision allow --audit-failure
python3 examples/s04-hooks/user_code/verify_practice.py

输入	程序是否运行	返回给模型的运行结果	宿主额外信息
普通允许	是	退出码 0、输出 42	请求与结果记录
维护阻止	否	带 run_1 的未执行反馈	请求与拒绝记录
审计故障	是，仍只有一次	退出码 0、输出 42	后置审计警告

标准库测试让真实临时程序每次执行都向标记文件追加一个字符，证明后置故障没有造成二次运行；同时验证前置异常、错误返回和权限拒绝不会生成标记。固定 SDK 1.13.0 的模拟 HTTP 测试核对下一次模型请求仍收到原调用 ID 的正确结果，警告没有混进工具协议。模型响应是替身，文件和进程是真实的；在线模型尚未调用，显式 --live 才会接入 API。

再试 --block-run --decision allow：所有普通回调都通过，也不能覆盖宿主禁止项。用重复工具请求触发调用上限，循环仍按原预算结束。本课没有新增 Stop 事件和自动续跑；只增加两个有明确契约的同步位置，就足以把维护检查、审计和原工具结果分别讲清楚。

资料核对于 2026-10-18：课程 Hooks提供生命周期扩展的学习路线；官方 Hooks 文档用于核对产品事件。本课的 before / after、故障策略及报告结构是本册 Python 设计，after 并非官方 PostToolUse 的同义事件。