第 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 立刻收工

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

本章完整示例代码

以下是本章涉及的完整示例文件，可以直接复制运行。

safety_guard.py — 安全防护（拦截危险命令）

import asyncio

from claude_agent_sdk import ClaudeAgentOptions, ClaudeSDKClient
from claude_agent_sdk.types import (
    AssistantMessage,
    HookContext,
    HookInput,
    HookJSONOutput,
    HookMatcher,
    ResultMessage,
    TextBlock,
)

# ============================================================
# 第一步：定义危险关键词列表
# ============================================================

DANGEROUS_PATTERNS = [
    "rm -rf",       # 递归强制删除
    "sudo",         # 提权操作
    "mkfs",         # 格式化磁盘
    "> /dev/",      # 写入设备文件
    "chmod 777",    # 全开权限
    "dd if=",       # 磁盘底层写入
    ":(){:|:&};:",  # Fork 炸弹
]


# ============================================================
# 第二步：编写 PreToolUse 钩子函数
# ============================================================

async def safety_guard(
    input_data: HookInput,
    tool_use_id: str | None,
    context: HookContext,
) -> HookJSONOutput:
    """检查 Bash 命令是否包含危险操作。

    逻辑很简单：
    - 遍历危险关键词列表
    - 命令里有任何一个关键词就拦截
    - 没有就放行
    """
    # 只关心 Bash 工具（虽然 matcher 已经限定了，但双重保险）
    if input_data["tool_name"] != "Bash":
        return {}

    command = input_data["tool_input"].get("command", "")
    print(f"  [安检门] 正在检查命令: {command}")

    # 逐个匹配危险关键词
    for pattern in DANGEROUS_PATTERNS:
        if pattern in command:
            print(f"  [安检门] 发现危险操作 '{pattern}'，已拦截!")
            return {
                # reason: 给 Claude 看的反馈，它会根据这个调整行为
                "reason": f"命令包含危险操作 '{pattern}'，已被安全策略拦截",
                # hookSpecificOutput: PreToolUse 专属的控制字段
                "hookSpecificOutput": {
                    "hookEventName": "PreToolUse",
                    "permissionDecision": "deny",
                    "permissionDecisionReason": f"安全策略禁止执行包含 '{pattern}' 的命令",
                },
            }

    # 没发现危险操作，放行
    print("  [安检门] 命令安全，放行")
    return {}


# ============================================================
# 第三步：配置 SDK 并运行
# ============================================================

async def main() -> None:
    # 创建配置，把钩子挂上去
    options = ClaudeAgentOptions(
        # 只允许 Claude 使用 Bash 工具
        allowed_tools=["Bash"],
        # 配置钩子
        hooks={
            "PreToolUse": [
                HookMatcher(
                    matcher="Bash",          # 只对 Bash 工具生效
                    hooks=[safety_guard],     # 使用我们写的安全检查函数
                ),
            ],
        },
    )

    async with ClaudeSDKClient(options=options) as client:
        # ---- 测试 1：安全命令（应该被放行）----
        print("=" * 50)
        print("测试 1: 安全命令")
        print("=" * 50)
        print("用户: 请执行 echo 'Hello, World!'")
        print()

        await client.query("请执行这个 bash 命令: echo 'Hello, World!'")
        async for msg in client.receive_response():
            if isinstance(msg, AssistantMessage):
                for block in msg.content:
                    if isinstance(block, TextBlock):
                        print(f"  Claude: {block.text}")
            elif isinstance(msg, ResultMessage):
                print(f"  [结果] 对话轮数: {msg.num_turns}")

        print()

        # ---- 测试 2：危险命令（应该被拦截）----
        print("=" * 50)
        print("测试 2: 危险命令")
        print("=" * 50)
        print("用户: 请执行 rm -rf /tmp/test_dir")
        print()

        await client.query("请执行这个 bash 命令: rm -rf /tmp/test_dir")
        async for msg in client.receive_response():
            if isinstance(msg, AssistantMessage):
                for block in msg.content:
                    if isinstance(block, TextBlock):
                        print(f"  Claude: {block.text}")
            elif isinstance(msg, ResultMessage):
                print(f"  [结果] 对话轮数: {msg.num_turns}")

    print()
    print("测试完成!")


if __name__ == "__main__":
    asyncio.run(main())

运行方式：

python examples/safety_guard.py

audit_logger.py — 审计日志

import asyncio
from datetime import datetime
from typing import Any

from claude_agent_sdk import ClaudeAgentOptions, ClaudeSDKClient
from claude_agent_sdk.types import (
    AssistantMessage,
    HookContext,
    HookInput,
    HookJSONOutput,
    HookMatcher,
    ResultMessage,
    TextBlock,
)

# ============================================================
# 第一步：定义审计日志存储
# ============================================================

# 用一个简单的列表来存放日志记录
# 生产环境中你可能会写到数据库或日志文件
audit_log: list[dict[str, Any]] = []


# ============================================================
# 第二步：编写 PreToolUse 钩子（记录调用前）
# ============================================================

async def log_tool_start(
    input_data: HookInput,
    tool_use_id: str | None,
    context: HookContext,
) -> HookJSONOutput:
    """记录工具调用开始。

    在工具执行之前，先把"谁要做什么"记下来。
    """
    record = {
        "timestamp": datetime.now().isoformat(),
        "event": "tool_start",
        "tool_name": input_data["tool_name"],
        "tool_input": input_data["tool_input"],
        "tool_use_id": tool_use_id,
        "session_id": input_data.get("session_id", "unknown"),
    }

    audit_log.append(record)
    print(f"  [审计] 工具调用开始: {record['tool_name']}")

    # 返回空字典 = 纯记录，不干预 Claude 的操作
    return {}


# ============================================================
# 第三步：编写 PostToolUse 钩子（记录调用后）
# ============================================================

async def log_tool_end(
    input_data: HookInput,
    tool_use_id: str | None,
    context: HookContext,
) -> HookJSONOutput:
    """记录工具调用结束和执行结果。

    工具执行完了，把结果也记下来。
    结果可能很长，截断到 200 字符。
    """
    tool_response = input_data.get("tool_response", "")
    # 把结果转成字符串并截断，避免日志过大
    response_str = str(tool_response)[:200]

    record = {
        "timestamp": datetime.now().isoformat(),
        "event": "tool_end",
        "tool_name": input_data["tool_name"],
        "tool_response_preview": response_str,
        "tool_use_id": tool_use_id,
    }

    audit_log.append(record)
    print(f"  [审计] 工具调用结束: {record['tool_name']} -> {response_str[:80]}...")

    # 同样不干预
    return {}


# ============================================================
# 第四步：打印日志摘要
# ============================================================

def print_audit_summary() -> None:
    """打印审计日志的汇总信息。"""
    print()
    print("=" * 60)
    print("审计日志摘要")
    print("=" * 60)

    if not audit_log:
        print("  (没有记录到任何工具调用)")
        return

    # 统计各工具的调用次数
    tool_counts: dict[str, int] = {}
    for record in audit_log:
        if record["event"] == "tool_start":
            name = record["tool_name"]
            tool_counts[name] = tool_counts.get(name, 0) + 1

    print(f"  总记录数: {len(audit_log)}")
    print(f"  工具调用次数:")
    for tool_name, count in tool_counts.items():
        print(f"    - {tool_name}: {count} 次")

    print()
    print("  详细记录:")
    print("  " + "-" * 56)
    for i, record in enumerate(audit_log, 1):
        event_type = "开始" if record["event"] == "tool_start" else "结束"
        time_str = record["timestamp"].split("T")[1][:8]  # 只取时分秒
        print(f"  {i}. [{time_str}] {record['tool_name']} ({event_type})")

        if record["event"] == "tool_start":
            # 显示输入参数
            tool_input = record.get("tool_input", {})
            input_preview = str(tool_input)[:60]
            print(f"     输入: {input_preview}")
        else:
            # 显示结果预览
            response = record.get("tool_response_preview", "")[:60]
            print(f"     结果: {response}")

    print("  " + "-" * 56)


# ============================================================
# 第五步：配置 SDK 并运行
# ============================================================

async def main() -> None:
    # 把两个钩子都挂上
    options = ClaudeAgentOptions(
        # 允许 Claude 使用 Bash 工具
        allowed_tools=["Bash"],
        hooks={
            # 工具调用前：记录开始
            "PreToolUse": [
                HookMatcher(
                    matcher=None,            # 匹配所有工具
                    hooks=[log_tool_start],
                ),
            ],
            # 工具调用后：记录结果
            "PostToolUse": [
                HookMatcher(
                    matcher=None,            # 匹配所有工具
                    hooks=[log_tool_end],
                ),
            ],
        },
    )

    print("开始与 Claude 对话（所有工具调用将被记录）...")
    print()

    async with ClaudeSDKClient(options=options) as client:
        # 让 Claude 执行几个命令，好让审计日志有东西
        print("用户: 请依次执行以下命令: echo 'hello', date, whoami")
        print()

        await client.query(
            "请依次执行以下 bash 命令，每个命令单独执行: "
            "echo 'hello world', date, whoami"
        )

        async for msg in client.receive_response():
            if isinstance(msg, AssistantMessage):
                for block in msg.content:
                    if isinstance(block, TextBlock):
                        print(f"  Claude: {block.text}")
            elif isinstance(msg, ResultMessage):
                print(f"  [结果] 轮数: {msg.num_turns}, "
                      f"耗时: {msg.duration_ms}ms")

    # 对话结束后，打印审计日志
    print_audit_summary()


if __name__ == "__main__":
    asyncio.run(main())

运行方式：

python examples/audit_logger.py

input_filter.py — 输入增强（自动注入上下文）

import asyncio
import platform
from datetime import datetime

from claude_agent_sdk import ClaudeAgentOptions, ClaudeSDKClient
from claude_agent_sdk.types import (
    AssistantMessage,
    HookContext,
    HookInput,
    HookJSONOutput,
    HookMatcher,
    ResultMessage,
    TextBlock,
)


# ============================================================
# 第一步：编写 UserPromptSubmit 钩子
# ============================================================

async def add_datetime_context(
    input_data: HookInput,
    tool_use_id: str | None,
    context: HookContext,
) -> HookJSONOutput:
    """自动给用户输入添加日期时间和环境上下文。

    每次用户提交问题时，这个钩子都会运行。
    它不会修改用户的原始输入，而是通过 additionalContext
    给 Claude 补充额外信息。
    """
    # 获取用户的原始输入
    user_prompt = input_data["prompt"]
    print(f"  [输入增强] 用户原始输入: {user_prompt}")

    # 构造上下文信息
    now = datetime.now()
    context_info = (
        f"当前时间: {now.strftime('%Y-%m-%d %H:%M:%S')} "
        f"({now.strftime('%A')})\n"
        f"操作系统: {platform.system()} {platform.release()}\n"
        f"Python 版本: {platform.python_version()}"
    )

    print(f"  [输入增强] 注入上下文: {context_info}")

    # 通过 additionalContext 把上下文信息传给 Claude
    # Claude 会看到这些信息，但用户不会
    return {
        "hookSpecificOutput": {
            "hookEventName": "UserPromptSubmit",
            "additionalContext": context_info,
        }
    }


# ============================================================
# 第二步：配置 SDK 并运行
# ============================================================

async def main() -> None:
    options = ClaudeAgentOptions(
        # 限制只回答一轮，不让 Claude 调用工具
        max_turns=1,
        hooks={
            "UserPromptSubmit": [
                HookMatcher(
                    # UserPromptSubmit 不涉及具体工具，matcher 必须是 None
                    matcher=None,
                    hooks=[add_datetime_context],
                ),
            ],
        },
    )

    async with ClaudeSDKClient(options=options) as client:
        # ---- 测试 1：问当前时间 ----
        print("=" * 50)
        print("测试 1: 问当前时间")
        print("=" * 50)
        print("用户: 现在是几月几号？星期几？几点了？")
        print()

        await client.query("现在是几月几号？星期几？几点了？")
        async for msg in client.receive_response():
            if isinstance(msg, AssistantMessage):
                for block in msg.content:
                    if isinstance(block, TextBlock):
                        print(f"  Claude: {block.text}")
            elif isinstance(msg, ResultMessage):
                print(f"  [结果] 对话轮数: {msg.num_turns}")

        print()

        # ---- 测试 2：问环境信息 ----
        print("=" * 50)
        print("测试 2: 问环境信息")
        print("=" * 50)
        print("用户: 我用的什么操作系统和 Python 版本？")
        print()

        await client.query("我现在用的什么操作系统？Python 是哪个版本？")
        async for msg in client.receive_response():
            if isinstance(msg, AssistantMessage):
                for block in msg.content:
                    if isinstance(block, TextBlock):
                        print(f"  Claude: {block.text}")
            elif isinstance(msg, ResultMessage):
                print(f"  [结果] 对话轮数: {msg.num_turns}")

    print()
    print("测试完成!")
    print()
    print("注意: Claude 本身不知道当前时间和你的系统信息。")
    print("它之所以能回答，是因为 UserPromptSubmit 钩子")
    print("在每次提问时自动注入了这些上下文。")


if __name__ == "__main__":
    asyncio.run(main())

运行方式：

python examples/input_filter.py

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                  # 输入增强：自动添加上下文

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

1. 钩子是什么？

2. 十种钩子事件

2.1 高频三剑客（必学）

PreToolUse — 工具调用前拦截

PostToolUse — 工具调用后处理

Stop — 对话结束时触发

2.2 中频事件（按需使用）

UserPromptSubmit — 用户输入预处理

Notification — 通知事件

PostToolUseFailure — 工具调用失败时

2.3 低频事件（了解即可）

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

3.1 input_data 里有什么？

3.2 返回值决定一切

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

拒绝工具调用（仅 PreToolUse）

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

修改工具输入（仅 PreToolUse）

添加额外上下文

停止整个对话

显示警告信息

4. HookMatcher — 精准匹配

matcher 字符串规则

超时设置

一个事件可以挂多个钩子

完整配置示例

5. 实战场景

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

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

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

6. 常见问题

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

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

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

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

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

本章完整示例代码

safety_guard.py — 安全防护（拦截危险命令）

audit_logger.py — 审计日志

input_filter.py — 输入增强（自动注入上下文）

7. 小结

本章文件清单

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

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