控制执行：hooks、流式、中断、结构化输出 · Claude Agent SDK 实战

不改循环，也能管住它。

三个我遇到过的场面：

Agent 说“修复完成”，但根本没跑测试。你问它，它说“我改了代码，应该没问题”。
它把客户的手机号完整念了出来，还写进了总结。
用户点了“停止”，进程还在跑；或者停了之后，历史里留下一个没有结果的工具调用。

这些不是提示词问题，是控制问题。

提示词能“建议”它跑测试、打码、停下来；但不能“保证”。要保证，得在 Harness 层动手。

这一章讲四个 Hook 配方，外加流式、中断、结构化输出。

15.1 Hook 是 Harness 的控制面

Hook 是 CLI 在关键时刻回调你的 Python 函数。常用的七个：

事件	什么时候	你能做什么
`PreToolUse`	工具执行前	拦、放、问；改输入
`PostToolUse`	工具执行后	改输出（脱敏）、审计
`Stop`	模型想结束	不许停，给理由让它继续
`SubagentStop`	子 Agent 结束	审计、质量检查
`PreCompact`	压缩前	存档、写进度
`UserPromptSubmit`	用户消息进来	注入当前状态/规则
`PermissionRequest`	需要审批时	接你自己的审批流

这一章只讲最有用的四个：Stop、PostToolUse、PreToolUse、UserPromptSubmit。

15.2 配方一：Stop —— 没跑测试不许结束

这是 coding agent 最需要的 Hook。

思路：在 PostToolUse 里记录“有没有跑过测试”；在 Stop 里检查，没有就 block。

python

STATE = {"tests_ran": False}


async def on_post(inp, tool_use_id, ctx):
    if inp.get("tool_name") == "Bash":
        cmd = str(inp.get("tool_input", {}).get("command", ""))
        if "unittest" in cmd or "pytest" in cmd:
            STATE["tests_ran"] = True
    return {}


async def on_stop(inp, tool_use_id, ctx):
    if inp.get("stop_hook_active"):
        return {}          # 已经被拦过一次，放行，否则会无限循环
    if not STATE["tests_ran"]:
        return {
            "decision": "block",
            "reason": "You have not run the tests. Run `python3 -m unittest discover -s tests -v` before finishing.",
        }
    return {}

两个细节：

stop_hook_active 必须判断。 它是“这次 Stop 已经被拦过一次”的标志；不判断，模型会反复触发 Stop，无限循环。
理由要具体。 不要只说“再检查一下”；告诉它跑哪条命令。模型下一轮才知道怎么改。

Stop 的输出是 {"decision": "block", "reason": ...}。CLI 会把 reason 作为反馈给模型，让它继续。

还有一个坑：在某些基线里，output_format 和 Stop Hook 一起用时，Stop 的 block 可能被忽略。升级后要复验，不要假设两者一定能叠加。

更重要的是：Stop Hook 和权限配置必须一致。

我有一个真实的反例：Stop 要求“跑过测试才许结束”，但 allowed_tools 里没有 Bash，测试命令被权限拦住。模型没法满足 Stop，又不敢结束，只能反复试；12 轮后被 error_max_turns 截断。

Stop 是“要求”，权限是“能力”。你要求它做一件它做不到的事，它不会变通，只会撞墙。上线前把这两个配置放在一起检查。

15.3 配方二：PostToolUse —— 脱敏

工具结果里可能有手机号、密钥、内部 URL。你不想让它们进上下文，更不想让它们进日志。

PostToolUse 可以改工具输出：

python

PHONE, MASK = "13800001234", "138****1234"


def mask(value):
    if isinstance(value, str):
        return value.replace(PHONE, MASK)
    if isinstance(value, list):
        return [mask(v) for v in value]
    if isinstance(value, dict):
        return {k: mask(v) for k, v in value.items()}
    return value


async def on_post(inp, tool_use_id, ctx):
    if PHONE not in str(inp.get("tool_response", "")):
        return {}
    return {
        "hookSpecificOutput": {
            "hookEventName": "PostToolUse",
            "updatedToolOutput": mask(inp.get("tool_response")),
        }
    }

关键点：

updatedToolOutput 要和原输出同形。 原输出是 dict，你就返回 dict；是 list，就返回 list。形状不对，CLI 可能拒绝。
先脱敏，再审计。 日志里也不该有原始手机号；不要“先记下来再打码”。
异步日志用 async_。 如果只是记日志、不影响结果，返回 {"async_": True, "asyncTimeout": 5000}，不要挡住主循环。

15.4 配方三：PreToolUse —— 拦危险命令

PreToolUse 在工具执行前触发，可以拦、放、问：

python

async def on_pre(inp, tool_use_id, ctx):
    if inp.get("tool_name") == "Bash":
        cmd = str(inp.get("tool_input", {}).get("command", ""))
        if "rm -rf" in cmd:
            return {
                "hookSpecificOutput": {
                    "hookEventName": "PreToolUse",
                    "permissionDecision": "deny",
                    "permissionDecisionReason": "rm -rf is not allowed; use a scoped delete.",
                }
            }
    return {}

permissionDecision 可以是 allow / deny / ask；permissionDecisionReason 会回给模型。

为什么用 Hook 而不是只用 can_use_tool？

Hook 在权限模式之前，bypassPermissions 也绕不过；
Hook 可以看到所有工具调用，包括被 allowed_tools 自动放行的；
Hook 可以做“按命令文本”的拦截，比 Bash 白名单更细。

但 Hook 不是沙箱。它只能看文本；rm -rf 换个写法、编码一下，就可能绕过。真正的隔离还是容器/沙箱；Hook 是“第一道文本闸门”。

15.5 配方四：UserPromptSubmit —— 注入状态

用户每发一条消息，你可以在它进入模型之前补一段上下文：

python

async def on_prompt(inp, tool_use_id, ctx):
    return {
        "hookSpecificOutput": {
            "hookEventName": "UserPromptSubmit",
            "additionalContext": "Project rule: do not edit tests; run tests before claiming done.",
        }
    }

适合放：

当前日期/分支/环境；
progress.md 的摘要；
当前任务的约束。

不适合放：

整个 README；
每次都一样的长文档（放 CLAUDE.md 或 Skill）；
敏感信息。

additionalContext 每一轮都会进上下文，按“每轮都付钱”的标准控制长度。

15.6 Hook 的成本

Hook 是同步回调：CLI 等你返回，才继续。

所以：

一个 2 秒的 Hook，会让每次工具调用多 2 秒。 十个工具调用就是 20 秒。
能异步的日志，异步化。 async_ 不阻塞主循环。
能静态规则挡的，不要走 Hook。 disallowed_tools 比 Hook 便宜。
Hook 里不要做重活。 网络请求、大文件扫描，尽量放到后台。

Hook 是“控制面”，不是“计算面”。它的价值是“在正确的时间点做一个小决定”，不是“替模型干活”。

15.7 流式：首字时延

默认情况下，你拿到的是完整消息。要流式，开 include_partial_messages=True：

python

options = ClaudeAgentOptions(
    include_partial_messages=True,
    # 其他 options
)


async def main():
    async for msg in query(...):
        if isinstance(msg, StreamEvent):
            # msg.event 里有 text_delta / content_block_delta 等
            ...

流式不减少总时延，也不减少成本；它改善的是首字时延和“用户能看到进度”的体验。

适合：聊天、长回答、需要“打字机效果”的 UI。不适合：批处理、CI、只看最终结果的场景。

15.8 中断：停止不是杀进程

ClaudeSDKClient 有 interrupt()：

python

async def stop(client):
    await client.interrupt()

它是协作式的：

发出中断请求；当前轮的结果消息仍然会到来；
不是 kill -9，不会瞬间清空；
已经执行的工具调用、已经改的文件，不会自动回滚；
中断后历史仍然合法，可以继续下一轮。

同理，运行期还能 set_model() / set_permission_mode()；但换工具集要重开进程（第 10 章）。

如果你要“停下来并回滚”，需要文件检查点 + rewind_files()，或者自己保存 diff。中断只负责“停”，不负责“撤销”。

15.9 结构化输出：Schema 是契约

output_format 可以让模型按 JSON Schema 输出：

python

options = ClaudeAgentOptions(
    output_format={
        "type": "json_schema",
        "schema": {
            "type": "object",
            "properties": {
                "summary": {"type": "string"},
                "files": {"type": "array", "items": {"type": "string"}},
            },
            "required": ["summary", "files"],
        },
    },
)

适合：

下游要解析的固定结构；
需要字段校验的工单/报告；
多步流程里的中间结果。

不适合：

结构很深、字段很多（拆成多步，别让一次输出承担全部）；
需要自由文本解释的场景；
你还没想清楚 schema 的场景（先跑几次，再定 schema）。

Schema 是契约，不是保证。你仍然要在代码里校验：字段在不在、类型对不对、业务上是否合法。

15.10 跑一遍

完整代码在 code/sdkagent/ch15_hooks.py，包含四个配方：UserPromptSubmit、PreToolUse（拦 rm -rf）、PostToolUse（脱敏手机号）、Stop（没跑测试不许结束）。

bash

python3 -m pip install "claude-agent-sdk==0.2.168"
export ANTHROPIC_API_KEY="..."
cd code
python3 sdkagent/ch15_hooks.py

2026-10-21 的最终成功运行（SDK 0.2.168 + CLI 2.1.296 + deepseek-flash）：

text

[tool] Bash {'command': 'ls -la && find . -type f -name "*.py" | head -50'}
[tool] Bash {'command': 'python3 -m unittest discover -s tests -v 2>&1 | tail -60'}
[tool] Read {'file_path': '.../taskflow/core.py'}
[tool] Read {'file_path': '.../tests/test_core.py'}
[tool] Edit {'file_path': '.../taskflow/core.py', 'new_string': '    return max(delta, 0)'}
[tool] Bash {'command': 'python3 -m unittest discover -s tests -v'}
[result] success

结果：它跑了测试、改了实现、又跑测试，2/2 通过；Stop Hook 放行。

这次运行没有触发 PreToolUse（没有 rm -rf）和 PostToolUse（没有手机号），所以那两个配方的实际效果没有被这次任务验证到；它们仍在代码里，需要单独构造用例。

失败的那次也值得写：

第一次跑的时候，allowed_tools=["Read","Glob","Grep"]，Bash 不在里面；Stop Hook 又要求“跑过测试”。模型反复尝试跑测试（ls、find、python3 -m unittest、dangerouslyDisableSandbox），但测试命令始终过不了权限；12 轮后 error_max_turns。

这不是 Stop Hook 的 bug，是“要求”和“能力”不一致。把 Bash 放进白名单（或者用 can_use_tool 只放行测试命令）之后，才成功。

说明：Hook 的触发顺序、updatedToolOutput 的接受格式、Stop 与 output_format 的叠加行为，都以你的版本和实测为准。这次没有打印 num_turns / total_cost_usd；别把“没打印”当成“没花钱”。

15.11 下一章：SDK 的边界

Hook 能管住很多事，但它管不了“模型选错工具”“上下文不够”“没有评测”。

第 16 章把第 2 章的“三层差距”拿出来，用前面这些章的机制填满：SDK 替你做了什么、你还得做什么、什么时候该回到 tiny/裸 API。

本章代码 四个 Hook 配方：https://github.com/kevinkelin/claude-agent-sdk-book/blob/master/code/sdkagent/ch15_hooks.py 说明：本章代码未在本机运行（环境没装 SDK/凭据）；Hook 事件/返回格式以官方文档和你的实测为准。