第 6 章：钩子系统与事件驱动

Claude 很强，但你不能让它想干啥就干啥。这一章，我们给 Claude 装上"安检门"。

1. 钩子是什么？

想象一下机场安检。你（乘客）要登机，但在登机前必须过安检门。安检员可以做三件事：

放行 — 没问题，过去吧
拦截 — 对不起，这个不能带
修改 — 这把小刀没收，其他东西可以带走

Claude 的钩子（Hook）就是这个"安检门"。每当 Claude 想做点什么（比如执行 Bash 命令、写文件），它都得先过你设的钩子。你可以放行、拦截、甚至偷偷改掉它要做的事。

用户提问 --> Claude 思考 --> [钩子检查] --> 执行工具 --> [钩子检查] --> 返回结果
                              ^                          ^
                              |                          |
                         PreToolUse                PostToolUse
                        (执行前检查)              (执行后检查)

核心价值：你不需要信任 Claude 会做对的事，你只需要确保它不能做错的事。

2. 十种钩子事件

SDK 支持 10 种钩子事件。别被数量吓到，日常用到的就前三个。

2.1 高频三剑客（必学）

PreToolUse — 工具调用前拦截

最常用的钩子，没有之一。

Claude 每次想调用工具（Bash、Write、Edit 等），都会先触发这个事件。你可以在这里： - 检查命令是否安全 - 修改工具输入 - 直接拒绝执行

from claude_agent_sdk.types import HookContext, HookInput, HookJSONOutput


async def pre_tool_check(
    input_data: HookInput,
    tool_use_id: str | None,
    context: HookContext,
) -> HookJSONOutput:
    # input_data 里有这些关键字段：
    # - tool_name: 工具名，比如 "Bash"、"Write"、"Edit"
    # - tool_input: 工具参数，比如 {"command": "ls -la"}
    # - tool_use_id: 本次调用的唯一标识
    tool_name = input_data["tool_name"]
    tool_input = input_data["tool_input"]

    print(f"Claude 想使用 {tool_name}，参数: {tool_input}")

    # 返回空字典 = 放行，什么都不干预
    return {}

PostToolUse — 工具调用后处理

工具执行完了，结果出来了，你可以在这里： - 检查执行结果是否正常 - 给 Claude 添加额外上下文（"这个结果看起来有问题，请小心"） - 记录日志

async def post_tool_check(
    input_data: HookInput,
    tool_use_id: str | None,
    context: HookContext,
) -> HookJSONOutput:
    # PostToolUse 比 PreToolUse 多一个字段：
    # - tool_response: 工具的执行结果
    tool_name = input_data["tool_name"]
    tool_response = input_data.get("tool_response", "")

    # 如果结果里有 "error"，给 Claude 一个提示
    if "error" in str(tool_response).lower():
        return {
            "hookSpecificOutput": {
                "hookEventName": "PostToolUse",
                "additionalContext": "上一步出错了，请检查后再继续。",
            }
        }

    return {}

Stop — 对话结束时触发

Claude 认为任务完成、准备收工时触发。你可以： - 做收尾清理工作 - 检查 Claude 的最终输出是否符合要求 - 阻止它停下来，让它继续干活

async def on_stop(
    input_data: HookInput,
    tool_use_id: str | None,
    context: HookContext,
) -> HookJSONOutput:
    print("Claude 准备结束对话了")

    # 返回 continue_: False 可以真正停止对话
    # 返回空字典则放行（让 Claude 正常结束）
    return {}

2.2 中频事件（按需使用）

UserPromptSubmit — 用户输入预处理

用户提交提示词时触发。适合在用户输入的基础上自动添加上下文。

async def enrich_prompt(
    input_data: HookInput,
    tool_use_id: str | None,
    context: HookContext,
) -> HookJSONOutput:
    # input_data["prompt"] 就是用户的原始输入
    user_prompt = input_data["prompt"]
    print(f"用户说: {user_prompt}")

    # 给 Claude 注入额外上下文
    return {
        "hookSpecificOutput": {
            "hookEventName": "UserPromptSubmit",
            "additionalContext": "当前项目使用 Python 3.12，框架是 FastAPI。",
        }
    }

Notification — 通知事件

Claude 发出通知时触发（比如长任务的进度提示）。

async def on_notification(
    input_data: HookInput,
    tool_use_id: str | None,
    context: HookContext,
) -> HookJSONOutput:
    # input_data["message"] 是通知内容
    # input_data["notification_type"] 是通知类型
    print(f"通知: {input_data['message']}")
    return {}

PostToolUseFailure — 工具调用失败时

工具执行出错了（不是结果里有 error，而是工具本身崩了）。

async def on_tool_failure(
    input_data: HookInput,
    tool_use_id: str | None,
    context: HookContext,
) -> HookJSONOutput:
    # input_data["error"] 是错误信息
    error_msg = input_data["error"]
    print(f"工具 {input_data['tool_name']} 执行失败: {error_msg}")

    return {
        "hookSpecificOutput": {
            "hookEventName": "PostToolUseFailure",
            "additionalContext": "工具崩溃了，请换一种方式尝试。",
        }
    }

2.3 低频事件（了解即可）

事件	触发时机	典型用途
`SubagentStart`	子代理启动时	记录子代理信息
`SubagentStop`	子代理结束时	检查子代理的执行结果
`PreCompact`	上下文压缩前	自定义压缩指令
`PermissionRequest`	权限请求时	自定义权限决策逻辑

这几个事件在复杂的多代理场景下才会用到，初学阶段不需要关心。

3. 钩子回调函数怎么写？

每个钩子都是一个 async 函数，签名固定：

async def my_hook(
    input_data: HookInput,       # 输入数据，根据事件类型不同字段不同
    tool_use_id: str | None,     # 工具调用 ID（非工具事件时为 None）
    context: HookContext,        # 上下文（目前主要是预留字段）
) -> HookJSONOutput:            # 返回值决定你要做什么
    ...

3.1 input_data 里有什么？

input_data 是一个字典，不同事件类型包含不同字段。所有事件都有这些公共字段：

字段	说明
`session_id`	会话 ID
`transcript_path`	对话记录路径
`cwd`	当前工作目录
`hook_event_name`	事件名称（跟你注册的钩子类型一致）

工具相关事件（PreToolUse / PostToolUse）还会额外包含：

字段	说明
`tool_name`	工具名称，如 `"Bash"`、`"Write"`
`tool_input`	工具输入参数，如 `{"command": "ls"}`
`tool_use_id`	本次工具调用的唯一 ID
`tool_response`	工具执行结果（仅 PostToolUse）

3.2 返回值决定一切

钩子的返回值是一个字典，不同的返回值意味着不同的操作。这是最关键的部分，务必记住：

返回 `{}` — 什么都不做，放行

async def passthrough_hook(input_data, tool_use_id, context):
    # 最简单的钩子：啥也不干
    return {}

Claude 的操作不受任何影响，该干嘛干嘛。

拒绝工具调用（仅 PreToolUse）

async def deny_hook(input_data, tool_use_id, context):
    return {
        "hookSpecificOutput": {
            "hookEventName": "PreToolUse",
            "permissionDecision": "deny",
            "permissionDecisionReason": "安全策略不允许此操作",
        }
    }

Claude 的这次工具调用会被拒绝。Claude 会收到拒绝原因，然后它可能会尝试换一种方式。

明确允许工具调用（仅 PreToolUse）

async def allow_hook(input_data, tool_use_id, context):
    return {
        "hookSpecificOutput": {
            "hookEventName": "PreToolUse",
            "permissionDecision": "allow",
            "permissionDecisionReason": "已通过安全检查",
        }
    }

跳过默认的权限检查，直接放行。适合你已经验证过安全性的场景。

修改工具输入（仅 PreToolUse）

async def modify_input_hook(input_data, tool_use_id, context):
    # 比如，自动给所有 Bash 命令加上 timeout 前缀
    original_command = input_data["tool_input"].get("command", "")
    return {
        "hookSpecificOutput": {
            "hookEventName": "PreToolUse",
            "permissionDecision": "allow",
            "updatedInput": {
                "command": f"timeout 30 {original_command}",
            },
        }
    }

偷偷改掉 Claude 要执行的命令，Claude 不会知道。

添加额外上下文

async def add_context_hook(input_data, tool_use_id, context):
    return {
        "hookSpecificOutput": {
            "hookEventName": "PostToolUse",
            "additionalContext": "注意：这个文件上次被修改是 3 天前。",
        }
    }

在工具执行后给 Claude 补充信息，帮助它做出更好的判断。

停止整个对话

async def emergency_stop(input_data, tool_use_id, context):
    return {
        "continue_": False,
        "stopReason": "检测到危险操作，紧急停止",
    }

注意： 字段名是 continue_（带下划线），因为 continue 是 Python 保留字。 SDK 会自动把它转换成 CLI 能识别的 continue。

显示警告信息

async def warning_hook(input_data, tool_use_id, context):
    return {
        "systemMessage": "警告：检测到潜在风险操作",
        "reason": "这个命令可能会修改系统配置，请谨慎操作",
    }

systemMessage 显示给用户看，reason 是给 Claude 的反馈信息。

4. HookMatcher — 精准匹配

有了钩子函数，还需要告诉 SDK"这个钩子在什么时候对什么工具生效"。这就是 HookMatcher 的职责。

from claude_agent_sdk.types import HookMatcher

# 只匹配 Bash 工具
bash_matcher = HookMatcher(
    matcher="Bash",
    hooks=[my_hook_function],
)

# 匹配多个工具（用 | 分隔）
file_matcher = HookMatcher(
    matcher="Write|Edit|MultiEdit",
    hooks=[my_hook_function],
)

# 匹配所有工具（matcher=None 或不传）
all_matcher = HookMatcher(
    matcher=None,
    hooks=[my_hook_function],
)

matcher 字符串规则

写法	含义
`"Bash"`	只匹配 Bash 工具
`"Write"`	只匹配 Write 工具
`"Write\\|Edit"`	匹配 Write 或 Edit
`"Bash\\|Write\\|Edit\\|MultiEdit"`	匹配这四个中的任意一个
`None`	匹配所有（对于 UserPromptSubmit 等非工具事件必须用 None）

超时设置

# 给钩子设一个 10 秒的超时
matcher = HookMatcher(
    matcher="Bash",
    hooks=[my_slow_hook],
    timeout=10.0,  # 单位：秒
)

如果钩子函数在超时时间内没有返回，SDK 会跳过它。默认超时是 60 秒。

一个事件可以挂多个钩子

from claude_agent_sdk import ClaudeAgentOptions

options = ClaudeAgentOptions(
    hooks={
        "PreToolUse": [
            # 第一道关：对 Bash 的安全检查
            HookMatcher(matcher="Bash", hooks=[check_bash_safety]),
            # 第二道关：对所有工具的日志记录
            HookMatcher(matcher=None, hooks=[log_all_tools]),
        ],
        "PostToolUse": [
            HookMatcher(matcher=None, hooks=[review_output]),
        ],
    }
)

多个 HookMatcher 会按顺序执行。如果第一个拒绝了，后面的就不会执行了。

完整配置示例

把所有东西串起来，一个完整的配置长这样：

from claude_agent_sdk import ClaudeAgentOptions, ClaudeSDKClient
from claude_agent_sdk.types import HookMatcher

options = ClaudeAgentOptions(
    hooks={
        "PreToolUse": [
            HookMatcher(matcher="Bash", hooks=[check_bash_command]),
            HookMatcher(matcher="Write|Edit", hooks=[check_file_write]),
        ],
        "PostToolUse": [
            HookMatcher(matcher=None, hooks=[log_tool_result]),
        ],
        "UserPromptSubmit": [
            HookMatcher(matcher=None, hooks=[enrich_user_input]),
        ],
        "Stop": [
            HookMatcher(matcher=None, hooks=[on_conversation_end]),
        ],
    }
)

async with ClaudeSDKClient(options=options) as client:
    await client.query("帮我重构这个文件")
    async for msg in client.receive_response():
        ...  # 处理消息

5. 实战场景

光看 API 不过瘾，来看三个真实场景。

场景一：安全防护 — 拦截危险命令

完整代码见 examples/safety_guard.py

这是钩子最经典的用途。你的 Claude 代理在生产环境运行，万一它想 rm -rf / 怎么办？

核心思路：在 PreToolUse 钩子里检查 Bash 命令，发现危险关键词就拦截。

# 危险命令关键词
DANGEROUS_PATTERNS = ["rm -rf", "sudo", "mkfs", "> /dev/", "chmod 777"]

async def safety_guard(input_data, tool_use_id, context):
    if input_data["tool_name"] != "Bash":
        return {}

    command = input_data["tool_input"].get("command", "")

    for pattern in DANGEROUS_PATTERNS:
        if pattern in command:
            return {
                "reason": f"命令包含危险操作 '{pattern}'，已拦截",
                "hookSpecificOutput": {
                    "hookEventName": "PreToolUse",
                    "permissionDecision": "deny",
                    "permissionDecisionReason": f"安全策略禁止: {pattern}",
                },
            }

    # 安全命令，放行
    return {}

配置到 SDK：

options = ClaudeAgentOptions(
    hooks={
        "PreToolUse": [
            HookMatcher(matcher="Bash", hooks=[safety_guard]),
        ],
    }
)

场景二：审计日志 — 记录所有操作

完整代码见 examples/audit_logger.py

在合规场景下，你可能需要记录 Claude 的每一步操作。用 PreToolUse + PostToolUse 的组合，可以完整记录"做了什么"和"结果是什么"。

audit_log = []  # 审计日志列表

async def log_before(input_data, tool_use_id, context):
    """记录工具调用前的信息"""
    audit_log.append({
        "event": "tool_start",
        "tool": input_data["tool_name"],
        "input": input_data["tool_input"],
        "tool_use_id": tool_use_id,
    })
    return {}  # 纯记录，不干预

async def log_after(input_data, tool_use_id, context):
    """记录工具调用后的结果"""
    audit_log.append({
        "event": "tool_end",
        "tool": input_data["tool_name"],
        "response": str(input_data.get("tool_response", ""))[:200],
        "tool_use_id": tool_use_id,
    })
    return {}  # 纯记录，不干预

配置：

options = ClaudeAgentOptions(
    hooks={
        "PreToolUse": [
            HookMatcher(matcher=None, hooks=[log_before]),
        ],
        "PostToolUse": [
            HookMatcher(matcher=None, hooks=[log_after]),
        ],
    }
)

场景三：输入增强 — 自动添加上下文

完整代码见 examples/input_filter.py

每次用户提问时，自动在背后塞一些额外信息给 Claude。比如当前时间、项目信息、用户偏好等。

from datetime import datetime

async def add_datetime_context(input_data, tool_use_id, context):
    """自动给用户输入添加时间上下文"""
    now = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
    return {
        "hookSpecificOutput": {
            "hookEventName": "UserPromptSubmit",
            "additionalContext": f"当前时间: {now}",
        }
    }

配置：

options = ClaudeAgentOptions(
    hooks={
        "UserPromptSubmit": [
            HookMatcher(matcher=None, hooks=[add_datetime_context]),
        ],
    }
)

这样用户问"现在几点了"，Claude 就能准确回答——即使它本身不知道当前时间。

6. 常见问题

Q: 钩子函数必须是 async 的吗？

是的。 钩子签名是 Callable[..., Awaitable[HookJSONOutput]]，所以必须用 async def。如果你的逻辑不涉及异步操作，直接 return 就行，不需要 await 任何东西。

Q: 一个钩子能注册到多个事件吗？

可以，但通常不建议。不同事件的 input_data 结构不同，你需要在函数里判断 hook_event_name，代码容易变乱。建议每个事件写独立的钩子函数。

Q: 钩子执行出错了会怎样？

如果钩子函数抛出异常，SDK 会跳过这个钩子，Claude 的操作不受影响。建议在钩子里自行做好异常处理。

Q: 钩子能访问对话历史吗？

不能直接访问。但你可以通过 input_data 中的 transcript_path 字段找到对话记录文件。或者自己在钩子外部维护状态（比如用一个列表记录历史）。

Q: `continue_` 和 `permissionDecision: "deny"` 有什么区别？

permissionDecision: "deny" — 拒绝这一次工具调用，Claude 还能继续对话、尝试其他方式
continue_: False — 直接停止整个对话，Claude 立刻收工

一个是"这条路不通，换条路试试"，一个是"别走了，回家"。

7. 小结

这一章我们学了：

钩子是 Claude 的安检门 — 在 Claude 执行操作前后插入你的检查逻辑
10 种事件，3 个常用 — PreToolUse（执行前检查）、PostToolUse（执行后检查）、Stop（结束时触发）
返回值决定行为 — {} 放行、permissionDecision: "deny" 拒绝、continue_: False 停止
HookMatcher 做精准匹配 — 指定钩子只对特定工具生效
三大场景 — 安全防护、审计日志、输入增强

钩子系统让你从"祈祷 Claude 做对的事"变成"确保 Claude 不能做错的事"。在生产环境中，这是不可或缺的安全网。

本章文件清单

06-钩子系统与事件驱动/
  README.md                          # 你正在读的这个文件
  examples/
    safety_guard.py                  # 安全防护：拦截危险 Bash 命令
    audit_logger.py                  # 审计日志：记录所有工具调用
    input_filter.py                  # 输入增强：自动添加上下文