第12章：Hooks —— 在 Agent 干活的每个节点插一脚

一句话：用 Hooks 监控和控制 Agent 的每一步操作，就像在流水线上安排质检员。

本章目标

• 理解 Hooks 的概念和作用
• 掌握所有 Hook 类型的触发时机和用法
• 学会通过 Hook 返回值控制 Agent 行为
• 能独立实现审计日志、安全拦截、成本控制等实战场景

前置知识

• 需要先看完第4章（query() 函数）
• 需要先看完第5章（内置工具）
• 需要先看完第6章（权限控制）

12.1 Hooks 是什么？

先打个比方

想象一个工厂的生产线。产品从原料到成品，要经过好几个环节：切割、打磨、组装、喷漆、包装。在每个环节之间，工厂都会安排质检站 —— 检查上一步做得对不对，决定下一步要不要继续。

Agent 的工作流程也是一条"生产线"：

用户输入 → Agent 思考 → 选择工具 → 执行工具 → 拿到结果 → 再思考 → ... → 最终输出

Hooks 就是这条生产线上的"质检站"。你可以在每个关键节点插入自己的代码逻辑，做这些事：

和权限控制有什么区别？

第6章我们学了权限控制（Permissions），它是静态的规则 —— 事先定好"这个工具能用、那个不能用"。

Hooks 是动态的逻辑 —— 在运行时，根据具体的参数、上下文、当前状态来决定怎么处理。比如：

• 权限控制只能说"Bash 工具需要审批"
• Hooks 可以说"Bash 工具执行 ls 自动放行，执行 rm -rf 直接拒绝，执行其他命令弹窗询问"

两者配合使用，才能构建出真正安全的 Agent 系统。

12.2 所有 Hook 类型详解

Claude Agent SDK 提供了 12 种 Hook，覆盖了 Agent 执行的各个阶段。我们先看一张全景图：

                    ┌─── SessionStart (TS)
                    │
用户输入 ──→ UserPromptSubmit
                    │
                    ▼
            ┌── PreToolUse ──→ 执行工具 ──→ PostToolUse
            │                                    │
            │                              PostToolUseFailure (TS，失败时)
            │                                    │
            ├── PreToolUse ──→ 执行工具 ──→ PostToolUse
            │       ...（循环多次）...
            │
            ├── SubagentStart (TS) ──→ 子 Agent 工作 ──→ SubagentStop
            │
            ├── PreCompact（对话太长时触发压缩）
            │
            ├── PermissionRequest (TS，弹权限对话框时)
            │
            ├── Notification (TS，状态消息)
            │
            └── Stop ──→ SessionEnd (TS)

各 Hook 支持情况一览

下面我们逐个详解。

12.2.1 PreToolUse —— 工具执行前的"门卫"

触发时机： Agent 决定要调用某个工具，但还没真正执行的时候。

你能拿到什么数据：

• tool_name：要调用的工具名字（比如 Bash、Write、Edit）
• tool_input：工具的输入参数（比如 Bash 工具的 command 参数）
• hook_event_name：固定是 "PreToolUse"

你能做什么：

• 放行：返回 {} 或 permissionDecision: "allow"
• 拒绝：返回 permissionDecision: "deny"
• 询问用户：返回 permissionDecision: "ask"
• 修改参数：返回 updatedInput 改掉工具的输入
• 注入信息：返回 systemMessage 给 Agent 额外提示

这是最常用的 Hook，绝大多数场景都会用到它。

Python 示例：拦截对 .env 文件的写入

import asyncio
from claude_agent_sdk import query, ClaudeAgentOptions, HookMatcher


async def protect_env_files(input_data, tool_use_id, context):
    """保护 .env 文件不被修改"""
    file_path = input_data["tool_input"].get("file_path", "")
    file_name = file_path.split("/")[-1]

    if file_name == ".env":
        return {
            "hookSpecificOutput": {
                "hookEventName": input_data["hook_event_name"],
                "permissionDecision": "deny",
                "permissionDecisionReason": "禁止修改 .env 文件，里面有密钥！",
            }
        }

    # 其他文件正常放行
    return {}


async def main():
    async for message in query(
        prompt="帮我更新项目配置",
        options=ClaudeAgentOptions(
            hooks={
                "PreToolUse": [
                    HookMatcher(matcher="Write|Edit", hooks=[protect_env_files])
                ]
            }
        ),
    ):
        print(message)


asyncio.run(main())

TypeScript 示例：

import { query, ClaudeAgentOptions } from "@anthropic-ai/claude-agent-sdk";

async function protectEnvFiles(inputData: any, toolUseId: string, context: any) {
  const filePath = inputData.tool_input?.file_path ?? "";
  const fileName = filePath.split("/").pop();

  if (fileName === ".env") {
    return {
      hookSpecificOutput: {
        hookEventName: inputData.hook_event_name,
        permissionDecision: "deny" as const,
        permissionDecisionReason: "禁止修改 .env 文件，里面有密钥！",
      },
    };
  }

  return {};
}

for await (const message of query({
  prompt: "帮我更新项目配置",
  options: {
    hooks: {
      PreToolUse: [{ matcher: "Write|Edit", hooks: [protectEnvFiles] }],
    },
  },
})) {
  console.log(message);
}

重点理解 matcher： "Write|Edit" 是正则表达式，意思是"只在工具名是 Write 或 Edit 时触发这个 Hook"。如果不写 matcher，就对所有工具都触发。

12.2.2 PostToolUse —— 工具执行后的"验收员"

触发时机： 工具执行成功后。

你能拿到什么数据：

• tool_name：工具名字
• tool_input：工具的输入参数
• tool_result：工具的执行结果
• hook_event_name：固定是 "PostToolUse"

你能做什么：

• 记录日志（谁在什么时间调了什么工具，结果是什么）
• 注入系统消息（根据结果给 Agent 额外提示）
• 对结果做后处理

Python 示例：记录所有工具调用

import json
from datetime import datetime


async def log_tool_usage(input_data, tool_use_id, context):
    """记录每次工具调用的日志"""
    log_entry = {
        "timestamp": datetime.now().isoformat(),
        "tool": input_data.get("tool_name", "unknown"),
        "input": input_data.get("tool_input", {}),
        "result_preview": str(input_data.get("tool_result", ""))[:200],  # 截取前200字符
    }

    # 追加写入日志文件
    with open("agent_audit.log", "a") as f:
        f.write(json.dumps(log_entry, ensure_ascii=False) + "\n")

    return {}


# 使用：
# hooks={"PostToolUse": [HookMatcher(hooks=[log_tool_usage])]}

12.2.3 PostToolUseFailure —— 工具出错后的"救火员"（仅 TypeScript）

触发时机： 工具执行失败时（抛出异常、超时等）。

你能拿到什么数据：

• tool_name：工具名字
• tool_input：工具的输入参数
• error：错误信息

你能做什么：

• 记录错误日志
• 发告警通知
• 注入系统消息引导 Agent 换个方案

TypeScript 示例：

async function handleToolFailure(inputData: any, toolUseId: string, context: any) {
  console.error(`[工具失败] ${inputData.tool_name}: ${inputData.error}`);

  // 给 Agent 一个提示，让它换个思路
  return {
    systemMessage: `工具 ${inputData.tool_name} 执行失败了，请尝试其他方案。`,
  };
}

// 使用：
// hooks: { PostToolUseFailure: [{ hooks: [handleToolFailure] }] }

12.2.4 UserPromptSubmit —— 用户输入的"前台接待"

触发时机： 用户提交了一条消息，Agent 还没开始处理。

你能拿到什么数据：

• prompt：用户输入的原始文本
• hook_event_name：固定是 "UserPromptSubmit"

你能做什么：

• 过滤/清洗用户输入（比如去掉敏感信息）
• 注入额外上下文（比如当前时间、用户身份）
• 阻止某些输入

Python 示例：给每次对话自动注入当前时间

from datetime import datetime


async def inject_timestamp(input_data, tool_use_id, context):
    """在用户每次提问时，自动告诉 Agent 当前时间"""
    now = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
    return {
        "hookSpecificOutput": {
            "hookEventName": input_data["hook_event_name"],
            "additionalContext": f"当前时间是 {now}，请在回答中注意时效性。",
        }
    }


# 使用：
# hooks={"UserPromptSubmit": [HookMatcher(hooks=[inject_timestamp])]}

注意： UserPromptSubmit 的 matcher 会被忽略，因为它不涉及具体工具。不管你设不设 matcher，这个 Hook 都会在用户输入时触发。

12.2.5 Stop —— Agent 要下班时的"交接人"

触发时机： Agent 完成任务要停止运行了。

你能拿到什么数据：

• stop_reason：停止原因（比如 "end_turn"、"max_tokens" 等）
• hook_event_name：固定是 "Stop"

你能做什么：

• 保存会话状态
• 清理临时文件
• 发送通知
• 也可以返回 continue: true 让 Agent 继续工作（但要小心死循环！）

Python 示例：在 Agent 停止时保存状态

import json


async def save_session_state(input_data, tool_use_id, context):
    """Agent 停止时保存一些状态信息"""
    state = {
        "stop_reason": input_data.get("stop_reason", "unknown"),
        "session_id": context.get("session_id", ""),
    }

    with open("session_state.json", "w") as f:
        json.dump(state, f, ensure_ascii=False, indent=2)

    print(f"[Hook] Agent 停止了，原因: {state['stop_reason']}")
    return {}


# 使用：
# hooks={"Stop": [HookMatcher(hooks=[save_session_state])]}

12.2.6 SubagentStart —— 子 Agent 启动时的"入职登记"（仅 TypeScript）

触发时机： 主 Agent 派出一个子 Agent 去执行子任务时。

你能拿到什么数据：

• tool_use_id：父 Agent 发起这次子任务的调用 ID
• hook_event_name：固定是 "SubagentStart"

你能做什么：

• 追踪有多少子 Agent 在并行工作
• 注入额外上下文给子 Agent
• 记录子任务的开始时间

TypeScript 示例：

const activeSubagents = new Map<string, number>();

async function trackSubagentStart(inputData: any, toolUseId: string, context: any) {
  activeSubagents.set(toolUseId, Date.now());
  console.log(`[子Agent启动] ID: ${toolUseId}, 当前活跃子Agent数: ${activeSubagents.size}`);

  return {
    hookSpecificOutput: {
      hookEventName: inputData.hook_event_name,
      additionalContext: "请高效完成任务，避免不必要的工具调用。",
    },
  };
}

// 使用：
// hooks: { SubagentStart: [{ hooks: [trackSubagentStart] }] }

12.2.7 SubagentStop —— 子 Agent 结束时的"绩效考核"

触发时机： 子 Agent 完成任务后。

你能拿到什么数据：

• tool_use_id：对应的调用 ID（可以和 SubagentStart 中的对应起来）
• hook_event_name：固定是 "SubagentStop"

你能做什么：

• 收集子 Agent 的执行结果
• 统计执行耗时
• 检查结果质量

TypeScript 示例：

async function trackSubagentStop(inputData: any, toolUseId: string, context: any) {
  const startTime = activeSubagents.get(toolUseId);
  if (startTime) {
    const duration = Date.now() - startTime;
    console.log(`[子Agent完成] ID: ${toolUseId}, 耗时: ${duration}ms`);
    activeSubagents.delete(toolUseId);
  }
  return {};
}

// 使用：
// hooks: { SubagentStop: [{ hooks: [trackSubagentStop] }] }

12.2.8 PreCompact —— 对话压缩前的"档案管理员"

触发时机： 当对话内容太长，Agent 需要压缩（summarize）之前的对话时。

你能拿到什么数据：

• 当前的完整对话内容
• hook_event_name：固定是 "PreCompact"

你能做什么：

• 在压缩前保存完整的对话记录（压缩后原始内容就没了）
• 添加"一定要记住"的关键信息到系统消息中

Python 示例：在压缩前归档对话

import json
from datetime import datetime


async def archive_before_compact(input_data, tool_use_id, context):
    """在对话压缩前，把完整对话存档"""
    archive = {
        "timestamp": datetime.now().isoformat(),
        "event": "pre_compact",
        "message": "对话即将被压缩，此为压缩前的完整记录归档",
    }

    with open("conversation_archive.jsonl", "a") as f:
        f.write(json.dumps(archive, ensure_ascii=False) + "\n")

    # 提醒 Agent 一些关键信息不能忘
    return {
        "systemMessage": "对话即将被压缩。请确保关键的任务目标和已完成的步骤不会丢失。",
    }


# 使用：
# hooks={"PreCompact": [HookMatcher(hooks=[archive_before_compact])]}

12.2.9 PermissionRequest —— 权限弹窗的"审批人"（仅 TypeScript）

触发时机： 当 Agent 需要弹出权限确认对话框时。

你能做什么：

• 自定义权限审批逻辑（比如自动审批某些操作，或者发到 Slack 让人审批）

12.2.10 SessionStart —— 会话开始的"开工仪式"（仅 TypeScript）

触发时机： 一个新的 Agent 会话初始化时。

你能做什么：

• 初始化日志系统
• 设置遥测/监控
• 注入初始上下文

12.2.11 SessionEnd —— 会话结束的"收工清扫"（仅 TypeScript）

触发时机： Agent 会话结束销毁时。

你能做什么：

• 清理临时资源（文件、连接等）
• 发送统计报告
• 保存最终状态

12.2.12 Notification —— Agent 的"广播喇叭"（仅 TypeScript）

触发时机： Agent 发出状态消息时（比如"正在搜索文件..."、"分析完成"等）。

你能做什么：

• 把状态消息转发到 Slack、钉钉、PagerDuty 等
• 在 Web 界面上显示实时进度

TypeScript 示例：

async function forwardNotification(inputData: any, toolUseId: string, context: any) {
  const message = inputData.message || "";
  console.log(`[Agent 状态] ${message}`);

  // 可以转发到 Slack
  // await sendToSlack(message);

  return {};
}

// 使用：
// hooks: { Notification: [{ hooks: [forwardNotification] }] }

12.3 Hook 的返回值详解

Hook 回调函数的返回值决定了 Agent 接下来怎么行动。返回一个空对象 {} 表示"啥也不干，正常继续"。如果你想干预 Agent 的行为，就需要在返回值里填入特定的字段。

返回值分两层：顶层字段和 hookSpecificOutput 内部字段。

顶层字段

这些字段适用于所有类型的 Hook：

hookSpecificOutput 内部字段

这些字段需要放在 hookSpecificOutput 对象里：

返回值示例大全

示例1：直接放行（最简单）

async def allow_everything(input_data, tool_use_id, context):
    return {}

示例2：拒绝操作

async def deny_operation(input_data, tool_use_id, context):
    return {
        "hookSpecificOutput": {
            "hookEventName": input_data["hook_event_name"],
            "permissionDecision": "deny",
            "permissionDecisionReason": "此操作已被安全策略禁止",
        }
    }

示例3：允许但修改参数

async def force_safe_mode(input_data, tool_use_id, context):
    """把所有 Bash 命令都加上 --dry-run 参数"""
    original_command = input_data["tool_input"].get("command", "")
    return {
        "hookSpecificOutput": {
            "hookEventName": input_data["hook_event_name"],
            "permissionDecision": "allow",
            "updatedInput": {
                "command": f"{original_command} --dry-run"
            },
        }
    }

示例4：注入系统消息

async def inject_context(input_data, tool_use_id, context):
    return {
        "systemMessage": "提醒：当前处于生产环境，请格外小心操作。",
    }

示例5：停止 Agent 执行

async def stop_agent(input_data, tool_use_id, context):
    return {
        "continue": False,
        "stopReason": "Token 预算已用完，Agent 被强制停止",
    }

12.4 实战：审计日志系统

在生产环境中，你需要知道 Agent 干了什么。万一出了问题，审计日志就是你的"黑匣子"。

需求

• 记录每次工具调用的时间、工具名、输入参数
• 记录每次工具调用的结果
• 记录 Agent 的启动和停止
• 所有日志写入文件，方便事后审查

完整代码

import asyncio
import json
from datetime import datetime
from claude_agent_sdk import query, ClaudeAgentOptions, HookMatcher


class AuditLogger:
    """审计日志记录器"""

    def __init__(self, log_file="agent_audit.jsonl"):
        self.log_file = log_file
        self.call_count = 0

    def _write_log(self, event_type, data):
        """写入一条日志"""
        entry = {
            "timestamp": datetime.now().isoformat(),
            "event_type": event_type,
            "sequence": self.call_count,
            **data,
        }
        with open(self.log_file, "a", encoding="utf-8") as f:
            f.write(json.dumps(entry, ensure_ascii=False) + "\n")

    async def on_pre_tool_use(self, input_data, tool_use_id, context):
        """工具执行前记录"""
        self.call_count += 1
        self._write_log("PRE_TOOL_USE", {
            "tool_name": input_data.get("tool_name", "unknown"),
            "tool_input": input_data.get("tool_input", {}),
            "tool_use_id": tool_use_id,
        })
        # 放行，不做拦截
        return {}

    async def on_post_tool_use(self, input_data, tool_use_id, context):
        """工具执行后记录"""
        result = input_data.get("tool_result", "")
        # 结果可能很长，只记录前500字符
        result_preview = str(result)[:500]

        self._write_log("POST_TOOL_USE", {
            "tool_name": input_data.get("tool_name", "unknown"),
            "result_preview": result_preview,
            "tool_use_id": tool_use_id,
        })
        return {}

    async def on_stop(self, input_data, tool_use_id, context):
        """Agent 停止时记录"""
        self._write_log("AGENT_STOP", {
            "stop_reason": input_data.get("stop_reason", "unknown"),
            "total_tool_calls": self.call_count,
        })
        print(f"\n[审计] Agent 已停止。共调用工具 {self.call_count} 次。")
        print(f"[审计] 日志已保存到 {self.log_file}")
        return {}


async def main():
    logger = AuditLogger("my_agent_audit.jsonl")

    async for message in query(
        prompt="帮我查看当前目录下有哪些 Python 文件，并统计总行数",
        options=ClaudeAgentOptions(
            allowed_tools=["Bash", "Read", "Glob", "Grep"],
            hooks={
                # 所有工具执行前都记录
                "PreToolUse": [
                    HookMatcher(hooks=[logger.on_pre_tool_use])
                ],
                # 所有工具执行后都记录
                "PostToolUse": [
                    HookMatcher(hooks=[logger.on_post_tool_use])
                ],
                # Agent 停止时记录
                "Stop": [
                    HookMatcher(hooks=[logger.on_stop])
                ],
            },
        ),
    ):
        if hasattr(message, "content"):
            print(message.content, end="", flush=True)


asyncio.run(main())

日志输出示例

运行后，my_agent_audit.jsonl 文件里会有类似这样的内容：

{"timestamp":"2026-02-25T14:30:01.123","event_type":"PRE_TOOL_USE","sequence":1,"tool_name":"Glob","tool_input":{"pattern":"**/*.py"},"tool_use_id":"toolu_abc123"}
{"timestamp":"2026-02-25T14:30:01.456","event_type":"POST_TOOL_USE","sequence":1,"tool_name":"Glob","result_preview":"['main.py', 'utils.py', 'test_app.py']","tool_use_id":"toolu_abc123"}
{"timestamp":"2026-02-25T14:30:02.789","event_type":"PRE_TOOL_USE","sequence":2,"tool_name":"Bash","tool_input":{"command":"wc -l main.py utils.py test_app.py"},"tool_use_id":"toolu_def456"}
{"timestamp":"2026-02-25T14:30:03.012","event_type":"POST_TOOL_USE","sequence":2,"tool_name":"Bash","result_preview":"  42 main.py\n  18 utils.py\n  65 test_app.py\n 125 total","tool_use_id":"toolu_def456"}
{"timestamp":"2026-02-25T14:30:04.345","event_type":"AGENT_STOP","sequence":2,"stop_reason":"end_turn","total_tool_calls":2}

每一行都是一个 JSON 对象，方便后续用程序分析。

12.5 实战：安全拦截器

Agent 能执行 Bash 命令、写文件、删文件...如果不加限制，它可能会干出格的事。安全拦截器就是 Agent 的"保安"。

需求

• 阻止所有文件删除操作（rm 命令）
• 阻止危险的 Bash 命令（rm -rf、chmod 777、curl | bash 等）
• 阻止修改系统关键文件（/etc/、/usr/ 下的文件）
• 安全的操作自动放行，不用每次都问用户

完整代码

import asyncio
import re
from claude_agent_sdk import query, ClaudeAgentOptions, HookMatcher


# 危险命令黑名单（正则表达式）
DANGEROUS_PATTERNS = [
    r"\brm\s+(-[a-zA-Z]*f|-[a-zA-Z]*r|--force|--recursive)",  # rm -rf, rm -f, rm --force
    r"\brm\s+-[a-zA-Z]*\s+/",   # rm 删除根目录下的东西
    r"\bchmod\s+777\b",          # chmod 777
    r"\bcurl\b.*\|\s*\bbash\b",  # curl ... | bash （从网上下载并执行脚本）
    r"\bwget\b.*\|\s*\bbash\b",  # wget ... | bash
    r"\bmkfs\b",                 # 格式化磁盘
    r"\bdd\s+if=",               # dd 命令（磁盘操作）
    r">\s*/dev/sd",              # 写入磁盘设备
    r"\bshutdown\b",             # 关机
    r"\breboot\b",               # 重启
]

# 受保护的目录
PROTECTED_DIRS = ["/etc", "/usr", "/sys", "/boot", "/var/lib"]

# 安全命令白名单（这些命令自动放行）
SAFE_COMMANDS = [
    r"^\s*ls\b",       # 列出文件
    r"^\s*cat\b",      # 查看文件内容
    r"^\s*head\b",     # 查看文件开头
    r"^\s*tail\b",     # 查看文件末尾
    r"^\s*wc\b",       # 统计行数/字数
    r"^\s*echo\b",     # 打印
    r"^\s*pwd\b",      # 当前目录
    r"^\s*date\b",     # 日期
    r"^\s*whoami\b",   # 当前用户
    r"^\s*git\s+(log|status|diff|branch|show)\b",  # git 只读操作
]


def is_dangerous_command(command):
    """检查命令是否危险"""
    for pattern in DANGEROUS_PATTERNS:
        if re.search(pattern, command, re.IGNORECASE):
            return True, pattern
    return False, None


def is_safe_command(command):
    """检查命令是否在安全白名单中"""
    for pattern in SAFE_COMMANDS:
        if re.search(pattern, command.strip(), re.IGNORECASE):
            return True
    return False


def is_protected_path(file_path):
    """检查文件路径是否受保护"""
    for protected_dir in PROTECTED_DIRS:
        if file_path.startswith(protected_dir):
            return True
    return False


async def bash_security_hook(input_data, tool_use_id, context):
    """Bash 命令安全拦截"""
    command = input_data["tool_input"].get("command", "")

    # 检查是否是安全命令，是就直接放行
    if is_safe_command(command):
        return {
            "hookSpecificOutput": {
                "hookEventName": input_data["hook_event_name"],
                "permissionDecision": "allow",
            }
        }

    # 检查是否是危险命令
    dangerous, pattern = is_dangerous_command(command)
    if dangerous:
        return {
            "hookSpecificOutput": {
                "hookEventName": input_data["hook_event_name"],
                "permissionDecision": "deny",
                "permissionDecisionReason": f"命令被安全策略拦截（匹配规则: {pattern}）",
            }
        }

    # 其他命令交给用户决定
    return {
        "hookSpecificOutput": {
            "hookEventName": input_data["hook_event_name"],
            "permissionDecision": "ask",
        }
    }


async def file_write_security_hook(input_data, tool_use_id, context):
    """文件写入安全拦截"""
    file_path = input_data["tool_input"].get("file_path", "")

    if is_protected_path(file_path):
        return {
            "systemMessage": f"注意：{file_path} 位于受保护目录中，写入已被阻止。",
            "hookSpecificOutput": {
                "hookEventName": input_data["hook_event_name"],
                "permissionDecision": "deny",
                "permissionDecisionReason": f"受保护目录 {file_path}，禁止写入",
            },
        }

    # 非受保护路径，正常放行
    return {}


async def main():
    async for message in query(
        prompt="帮我清理项目中的临时文件和日志",
        options=ClaudeAgentOptions(
            allowed_tools=["Bash", "Read", "Write", "Edit", "Glob", "Grep"],
            hooks={
                "PreToolUse": [
                    # Bash 命令走安全拦截
                    HookMatcher(matcher="Bash", hooks=[bash_security_hook]),
                    # 文件写入走路径保护
                    HookMatcher(matcher="Write|Edit", hooks=[file_write_security_hook]),
                ],
            },
        ),
    ):
        if hasattr(message, "content"):
            print(message.content, end="", flush=True)


asyncio.run(main())

工作流程说明

Agent 想执行 "ls -la" → bash_security_hook 检查 → 在白名单中 → 自动放行
Agent 想执行 "rm -rf /tmp" → bash_security_hook 检查 → 匹配危险规则 → 直接拒绝
Agent 想执行 "npm install" → bash_security_hook 检查 → 不在白名单也不危险 → 弹窗让用户决定
Agent 想写入 "/etc/hosts" → file_write_security_hook 检查 → 受保护目录 → 直接拒绝
Agent 想写入 "/tmp/result.txt" → file_write_security_hook 检查 → 不受保护 → 正常放行

12.6 实战：成本控制器

Agent 调用 LLM 是按 Token 计费的。一个不小心，Agent 进入死循环或者处理超大文件，你的钱包就遭殃了。成本控制器帮你设一个"预算上限"。

需求

• 追踪 Agent 消耗的总 Token 数
• 当 Token 超过预算时，强制停止 Agent
• 在接近预算时发出警告
• 记录成本数据供事后分析

完整代码

import asyncio
import json
from datetime import datetime
from claude_agent_sdk import query, ClaudeAgentOptions, HookMatcher


class CostController:
    """成本控制器"""

    def __init__(self, max_tool_calls=50, warn_at=40):
        """
        参数：
            max_tool_calls: 最大允许的工具调用次数
            warn_at: 达到多少次时发出警告
        """
        self.max_tool_calls = max_tool_calls
        self.warn_at = warn_at
        self.tool_call_count = 0
        self.start_time = datetime.now()
        self.cost_log = []

    def _log(self, event, data):
        """记录成本事件"""
        entry = {
            "timestamp": datetime.now().isoformat(),
            "event": event,
            "tool_call_count": self.tool_call_count,
            **data,
        }
        self.cost_log.append(entry)

    async def on_pre_tool_use(self, input_data, tool_use_id, context):
        """每次工具调用前检查预算"""
        self.tool_call_count += 1
        tool_name = input_data.get("tool_name", "unknown")

        self._log("tool_call", {"tool_name": tool_name})

        # 超过预算上限，强制停止
        if self.tool_call_count > self.max_tool_calls:
            self._log("budget_exceeded", {
                "tool_name": tool_name,
                "limit": self.max_tool_calls,
            })
            return {
                "continue": False,
                "stopReason": f"成本控制：工具调用次数已达上限 {self.max_tool_calls} 次",
            }

        # 接近预算上限，注入警告
        if self.tool_call_count >= self.warn_at:
            remaining = self.max_tool_calls - self.tool_call_count
            return {
                "systemMessage": f"注意：你还剩 {remaining} 次工具调用机会，请尽快完成任务。",
            }

        return {}

    async def on_stop(self, input_data, tool_use_id, context):
        """Agent 停止时输出成本报告"""
        elapsed = (datetime.now() - self.start_time).total_seconds()

        report = {
            "total_tool_calls": self.tool_call_count,
            "elapsed_seconds": elapsed,
            "budget_limit": self.max_tool_calls,
            "budget_utilization": f"{self.tool_call_count / self.max_tool_calls * 100:.1f}%",
        }

        print("\n" + "=" * 50)
        print("成本报告")
        print("=" * 50)
        print(f"工具调用次数: {report['total_tool_calls']} / {report['budget_limit']}")
        print(f"预算使用率:   {report['budget_utilization']}")
        print(f"运行时长:     {report['elapsed_seconds']:.1f} 秒")
        print("=" * 50)

        # 保存详细日志
        with open("cost_report.json", "w", encoding="utf-8") as f:
            json.dump({
                "summary": report,
                "details": self.cost_log,
            }, f, ensure_ascii=False, indent=2)

        return {}


async def main():
    controller = CostController(max_tool_calls=20, warn_at=15)

    async for message in query(
        prompt="帮我分析当前项目的代码质量，包括代码行数、函数数量、注释比例",
        options=ClaudeAgentOptions(
            allowed_tools=["Bash", "Read", "Glob", "Grep"],
            hooks={
                "PreToolUse": [
                    HookMatcher(hooks=[controller.on_pre_tool_use])
                ],
                "Stop": [
                    HookMatcher(hooks=[controller.on_stop])
                ],
            },
        ),
    ):
        if hasattr(message, "content"):
            print(message.content, end="", flush=True)


asyncio.run(main())

成本报告输出示例

==================================================
成本报告
==================================================
工具调用次数: 12 / 20
预算使用率:   60.0%
运行时长:     18.3 秒
==================================================

12.7 Hook 组合使用

在实际项目中，你通常需要把多个 Hook 组合在一起使用。下面讲讲怎么组合，以及需要注意的事项。

同一事件挂多个 Hook

一个事件可以注册多个 HookMatcher，它们会按照注册顺序依次执行：

hooks={
    "PreToolUse": [
        # 第1个：安全拦截（优先级最高）
        HookMatcher(matcher="Bash", hooks=[bash_security_hook]),
        # 第2个：审计日志
        HookMatcher(hooks=[audit_logger.on_pre_tool_use]),
        # 第3个：成本控制
        HookMatcher(hooks=[cost_controller.on_pre_tool_use]),
    ],
}

执行顺序： 从上到下依次执行。如果第1个返回了 deny，后面的还会执行（用于日志记录），但最终的权限决策以"最严格的"为准。

不同事件组合

典型的组合方式是"PreToolUse + PostToolUse + Stop"三件套：

hooks={
    # 执行前：安全检查 + 审计记录
    "PreToolUse": [
        HookMatcher(matcher="Bash", hooks=[security.check_bash]),
        HookMatcher(matcher="Write|Edit", hooks=[security.check_file_write]),
        HookMatcher(hooks=[audit.on_pre_tool_use]),
    ],
    # 执行后：审计记录
    "PostToolUse": [
        HookMatcher(hooks=[audit.on_post_tool_use]),
    ],
    # 用户输入时：注入上下文
    "UserPromptSubmit": [
        HookMatcher(hooks=[context_injector.on_user_input]),
    ],
    # 停止时：保存状态 + 成本报告
    "Stop": [
        HookMatcher(hooks=[audit.on_stop]),
        HookMatcher(hooks=[cost_controller.on_stop]),
    ],
}

完整的"生产级" Hook 组合示例

下面是一个把审计、安全、成本控制全部组合在一起的完整例子：

import asyncio
from claude_agent_sdk import query, ClaudeAgentOptions, HookMatcher


# 复用前面定义的类
# from audit_logger import AuditLogger
# from security import bash_security_hook, file_write_security_hook
# from cost_controller import CostController


async def main():
    # 初始化各个组件
    audit = AuditLogger("production_audit.jsonl")
    cost = CostController(max_tool_calls=100, warn_at=80)

    async for message in query(
        prompt="分析项目代码并生成质量报告",
        options=ClaudeAgentOptions(
            allowed_tools=["Bash", "Read", "Write", "Glob", "Grep"],
            hooks={
                "PreToolUse": [
                    # 1. 安全拦截（最高优先级）
                    HookMatcher(matcher="Bash", hooks=[bash_security_hook]),
                    HookMatcher(matcher="Write|Edit", hooks=[file_write_security_hook]),
                    # 2. 成本控制
                    HookMatcher(hooks=[cost.on_pre_tool_use]),
                    # 3. 审计日志
                    HookMatcher(hooks=[audit.on_pre_tool_use]),
                ],
                "PostToolUse": [
                    HookMatcher(hooks=[audit.on_post_tool_use]),
                ],
                "Stop": [
                    HookMatcher(hooks=[audit.on_stop]),
                    HookMatcher(hooks=[cost.on_stop]),
                ],
            },
        ),
    ):
        if hasattr(message, "content"):
            print(message.content, end="", flush=True)


asyncio.run(main())

最佳实践

1. 安全拦截放最前面 —— 让安全检查第一个执行，危险操作直接拒绝，后面的 Hook 不需要处理
2. 日志记录放在最后 —— 确保不管前面的 Hook 做了什么决定，日志都能记下来
3. 用 matcher 缩小范围 —— 只给需要的工具注册 Hook，避免不必要的性能开销
4. Hook 回调要快 —— Hook 是同步阻塞的，执行太慢会拖慢 Agent 整体速度。如果需要做耗时操作（比如调外部 API），考虑使用异步方式并设置合理的 timeout
5. 不要在 Hook 里抛异常 —— Hook 抛异常会影响 Agent 的正常工作。用 try/except 包裹你的逻辑
6. 小心 UserPromptSubmit 中的无限循环 —— 如果你在 UserPromptSubmit Hook 里创建子 Agent，而子 Agent 又触发了同一个 Hook，就会形成无限循环。用一个标记位来防止递归

# 防止无限循环的写法
is_sub_agent = False

async def my_user_prompt_hook(input_data, tool_use_id, context):
    global is_sub_agent
    if is_sub_agent:
        return {}  # 子 Agent 不触发这个逻辑

    is_sub_agent = True
    try:
        # 你的逻辑...
        pass
    finally:
        is_sub_agent = False

    return {}

动手练习

练习1：实现审计系统 —— 记录 Agent 的所有操作

目标： 创建一个完整的审计日志系统。

要求：

1. 记录所有工具调用（工具名、输入参数、结果、耗时）
2. 记录用户的每次输入
3. 记录 Agent 的启动和停止
4. 日志格式为 JSONL（每行一个 JSON），方便后续分析
5. 在 Agent 停止时输出一个摘要统计（总共调了几个工具、最常用的工具是哪个、总耗时）

提示：

• 用一个类来管理状态（计数器、时间戳等）
• 同时注册 PreToolUse、PostToolUse、UserPromptSubmit、Stop 四个 Hook
• 用 tool_use_id 把 PreToolUse 和 PostToolUse 关联起来，计算每次调用的耗时

练习2：实现安全沙箱 —— 拦截危险的 Bash 命令

目标： 创建一个可配置的安全沙箱。

要求：

1. 支持自定义黑名单和白名单（通过配置文件或构造函数参数）
2. 白名单中的命令自动放行
3. 黑名单中的命令直接拒绝
4. 其他命令弹窗让用户决定
5. 限制 Agent 只能访问指定目录下的文件
6. 记录所有被拦截的操作

提示：

• 把安全策略做成一个类，构造函数接收白名单、黑名单、允许的目录列表
• 同时用 PreToolUse Hook 拦截 Bash 和 Write/Edit
• 被拦截时用 systemMessage 告诉 Agent 为什么被拒绝，引导它换个方式

练习3：实现成本控制器 —— 超过预算自动停止

目标： 创建一个生产级的成本控制系统。

要求：

1. 可以设置"最大工具调用次数"和"最大运行时间"两个维度的限制
2. 达到 80% 时发出警告（通过 systemMessage）
3. 达到 100% 时强制停止（通过 continue: false）
4. 停止时生成一份详细的成本报告（JSON 格式）
5. 报告包含：每种工具的调用次数、总运行时长、预算使用率

提示：

• 用 PreToolUse Hook 检查预算
• 用 Stop Hook 生成报告
• 运行时间限制可以用 datetime 模块实现
• 可以用 collections.Counter 统计每种工具的调用次数

本章小结

• Hooks 是 Agent 流水线上的"质检站"，让你在关键节点插入自定义逻辑
• 12 种 Hook 类型，覆盖了工具调用前后、用户输入、Agent 停止、子 Agent 生命周期等各个阶段
• Python SDK 支持 7 种（PreToolUse、PostToolUse、UserPromptSubmit、Stop、SubagentStop、PreCompact），TypeScript SDK 全部支持
• Hook 回调函数接收三个参数：input_data（事件数据）、tool_use_id（工具调用ID）、context（上下文）
• 返回值控制 Agent 行为：permissionDecision 控制放行/拒绝、updatedInput 修改参数、systemMessage 注入信息、continue: false 停止 Agent
• HookMatcher 的 matcher 参数是正则表达式，用于匹配工具名，只对 PreToolUse、PostToolUse 等工具相关 Hook 有效
• 三大实战场景：审计日志（安全合规）、安全拦截（防止误操作）、成本控制（预算管理）
• 组合使用时注意顺序：安全拦截放前面，日志记录放后面，小心无限循环

下一章预告

下一章我们要学多 Agent 协作 —— 一个人干不完就叫帮手。当任务太复杂、上下文太长、或者需要多种技能时，一个 Agent 忙不过来怎么办？答案是：派出一群 Agent，各司其职，协同作战。下一章会教你怎么用 Subagent 搭建"Agent 团队"，实现串行、并行、分工等协作模式。