第7课：监控AI的一举一动 —— Hooks 钩子系统
本课目标
多个 AI 在后台搜索、写文件、跑代码——你怎么知道它们都在干嘛？有没有乱来？这就需要 Hooks（钩子） 来监控和控制。
什么是 Hook？
Hook 就是你插在 AI 工作流程中的"监控摄像头"。
每当 Claude 要调用一个工具（比如 WebSearch、Write、Bash），系统会在调用前和调用后分别"通知"你的 Hook 函数。你可以：
- 看一眼就放行 — 记录日志
- 拦截并拒绝 — "不许执行这个命令！"
- 修改行为 — "你搜这个关键词不对，换一个"
graph TD
A["Claude 想搜索 AI 芯片"] --> B["PreToolUse Hook<br/>调用前: 你决定放不放行"]
B -- "放行" --> C["执行 WebSearch(AI 芯片)"]
C --> D["PostToolUse Hook<br/>调用后: 你看看结果怎么样"]
D --> E["结果返回给 Claude"]
最简 Hook：记录日志
"""
07_basic_hook.py
最简单的 Hook —— 记录所有工具调用
"""
import anyio
from claude_agent_sdk import (
ClaudeSDKClient, ClaudeAgentOptions, HookMatcher,
AssistantMessage, TextBlock
)
# Hook 函数：在工具调用前触发
async def log_tool_call(hook_input, tool_use_id, context):
"""每次 Claude 要用工具时，打印一条日志"""
tool_name = hook_input["tool_name"]
tool_input = hook_input["tool_input"]
print(f"📋 [日志] Claude 正在使用: {tool_name}")
print(f" 参数: {str(tool_input)[:100]}...")
return {} # 返回空字典 = 放行，不干预
# Hook 函数：在工具调用后触发
async def log_tool_result(hook_input, tool_use_id, context):
"""工具执行完后，记录结果"""
tool_name = hook_input["tool_name"]
print(f"✅ [完成] {tool_name} 执行完毕")
return {}
async def main():
options = ClaudeAgentOptions(
system_prompt="你是一个助手，会搜索信息。",
allowed_tools=["WebSearch", "Write"],
permission_mode="bypassPermissions",
max_turns=5,
hooks={
# PreToolUse: 工具调用前
"PreToolUse": [
HookMatcher(
matcher=None, # None 表示匹配所有工具
hooks=[log_tool_call]
),
],
# PostToolUse: 工具调用后
"PostToolUse": [
HookMatcher(
matcher=None,
hooks=[log_tool_result]
),
],
}
)
async with ClaudeSDKClient(options=options) as client:
await client.query("搜索一下最新的 Python 3.13 有什么新特性")
async for msg in client.receive_response():
if isinstance(msg, AssistantMessage):
for block in msg.content:
if isinstance(block, TextBlock):
print(f"\n💬 {block.text}\n")
anyio.run(main)
运行后你会看到：
📋 [日志] Claude 正在使用: WebSearch
参数: {'query': 'Python 3.13 新特性'}...
✅ [完成] WebSearch 执行完毕
📋 [日志] Claude 正在使用: WebSearch
参数: {'query': 'Python 3.13 new features 2025'}...
✅ [完成] WebSearch 执行完毕
💬 Python 3.13 带来了以下新特性...
每一步操作都清清楚楚！
Hook 进阶：拦截危险操作
光记录不够，有时候你得拦住 AI 的危险操作：
"""
07_safety_hook.py
安全钩子 —— 拦截危险命令
"""
import anyio
from claude_agent_sdk import (
ClaudeSDKClient, ClaudeAgentOptions, HookMatcher,
AssistantMessage, TextBlock
)
async def safety_check(hook_input, tool_use_id, context):
"""检查 Bash 命令，拦截危险操作"""
tool_name = hook_input["tool_name"]
tool_input = hook_input["tool_input"]
if tool_name != "Bash":
return {} # 不是 Bash 命令，放行
command = tool_input.get("command", "")
# 黑名单：这些命令绝对不能执行
dangerous_patterns = [
"rm -rf", # 删除文件
"sudo", # 提权
"chmod 777", # 改权限
"curl | bash", # 从网上下载并执行脚本
"DROP TABLE", # 删数据库表
]
for pattern in dangerous_patterns:
if pattern in command:
print(f"🚫 [拦截] 危险命令被阻止: {command}")
return {
"hookSpecificOutput": {
"hookEventName": "PreToolUse",
"permissionDecision": "deny",
"permissionDecisionReason": f"安全检查：命令包含危险模式 '{pattern}'",
}
}
print(f"✅ [放行] 安全命令: {command[:60]}...")
return {}
async def main():
options = ClaudeAgentOptions(
allowed_tools=["Bash"],
permission_mode="bypassPermissions",
max_turns=5,
hooks={
"PreToolUse": [
HookMatcher(matcher="Bash", hooks=[safety_check]),
# ↑ 只针对 Bash 工具触发
],
}
)
async with ClaudeSDKClient(options=options) as client:
# 安全命令 —— 会被放行
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(block.text)
anyio.run(main)
关键区别：
- return {} → 放行
- return {"hookSpecificOutput": {"permissionDecision": "deny", ...}} → 拦截
HookMatcher 的 matcher 参数
HookMatcher(matcher=None, hooks=[...]) # 匹配所有工具
HookMatcher(matcher="Bash", hooks=[...]) # 只匹配 Bash
HookMatcher(matcher="Write", hooks=[...]) # 只匹配 Write
你可以为不同工具设置不同的 Hook：
hooks = {
"PreToolUse": [
HookMatcher(matcher="Bash", hooks=[check_bash_safety]),
HookMatcher(matcher="Write", hooks=[check_write_path]),
HookMatcher(matcher=None, hooks=[log_all_tool_calls]),
],
}
研究平台的 Hook 实战：子Agent追踪器
先搞懂问题：为什么需要追踪？
想象你是一个公司老板，你让组长安排 3 个员工同时干活： - 研究员去搜资料 - 分析师去跑数据 - 报告员去写文档
你坐在办公室里，怎么知道每个人现在在干嘛？答案是：装一个监控系统，每个人每做一件事都打卡记录。
子Agent追踪器就是这个"打卡系统"。
追踪器长什么样？
# 简化版的子Agent追踪器
class SubagentTracker:
def __init__(self):
self.sessions = {} # 记录所有子Agent的活动
self._current_parent_id = None # 当前正在执行的子Agent
async def pre_tool_use_hook(self, hook_input, tool_use_id, context):
"""工具调用前：记录是哪个Agent在用什么工具"""
tool_name = hook_input["tool_name"]
# 跳过 Task 工具本身（那是组长在派活）
if tool_name == "Task":
return {}
# 找到当前是哪个子Agent在工作
if self._current_parent_id and self._current_parent_id in self.sessions:
session = self.sessions[self._current_parent_id]
agent_id = session["id"] # 比如 "RESEARCHER-1"
print(f"[{agent_id}] → {tool_name}")
# 记录这次工具调用
session["tool_calls"].append({
"tool": tool_name,
"input": hook_input["tool_input"],
"timestamp": "...",
})
return {}
async def post_tool_use_hook(self, hook_input, tool_use_id, context):
"""工具调用后：记录结果"""
# 记录输出、错误等信息
return {}
光有类还不行——完整的接线方式
上面只是定义了一个类，但它还没有被"接入"到 Claude 的工作流里。
在完成接线之前，有一个关键问题要先搞清楚——hook_input 里到底有什么字段？
先搞清楚 hook_input 里有什么
不同的 Hook 事件，hook_input 的内容是不一样的。你记不住没关系，写一个"探测 Hook"把它打印出来就行：
async def debug_hook(hook_input, tool_use_id, context):
"""开发时用的调试 Hook：把收到的所有信息都打印出来"""
print(f"🔍 hook_input 的内容: {hook_input}")
print(f"🔍 tool_use_id: {tool_use_id}")
print(f"🔍 context 的 keys: {list(context.keys()) if context else 'None'}")
return {}
把它挂到任何事件上，一运行就能看到实际传过来了什么数据。比如：
# 挂到 PreToolUse 上，你会看到类似这样的输出：
🔍 hook_input 的内容: {
"tool_name": "WebSearch",
"tool_input": {"query": "电池技术 2025"}
}
# 挂到 SubagentStart 上，你会看到类似这样的输出：
🔍 hook_input 的内容: {
"session_id": "abc123",
"prompt": "搜索电池技术的最新进展..."
}
结论：hook_input 的字段不是你猜的，是 SDK 定义好的。不确定有什么字段？打印一下就知道了。
不同事件的 hook_input 常见字段：
| Hook 事件 | hook_input 常见字段 | 说明 |
|---|---|---|
PreToolUse |
tool_name, tool_input |
要调用的工具名和参数 |
PostToolUse |
tool_name, tool_input, tool_output |
多了工具的返回结果 |
SubagentStart |
session_id, prompt |
子Agent 的会话 ID 和任务指令 |
SubagentStop |
session_id |
子Agent 的会话 ID |
⚠️ 实际的字段名以你的 SDK 版本为准。第一次用某个事件时，永远先用上面的 debug_hook 打印一下，确认字段名再写代码。 这是老手和新手的区别——老手不靠猜，靠打印。
完整的接线代码
知道了 hook_input 里有什么，现在可以写完整代码了：
"""
07_subagent_tracker_full.py
完整的子Agent追踪器使用示例
"""
import anyio
from claude_agent_sdk import (
ClaudeSDKClient, ClaudeAgentOptions, HookMatcher,
AssistantMessage, TextBlock, ToolUseBlock
)
# ============ 第一步：创建追踪器实例 ============
tracker = SubagentTracker()
# 子Agent计数器，用来生成 RESEARCHER-1、RESEARCHER-2 这样的编号
agent_counter = 0
# ============ 第二步：写额外的 Hook 来追踪子Agent的生命周期 ============
async def on_subagent_start(hook_input, tool_use_id, context):
"""当一个子Agent被组长派出去时触发"""
global agent_counter
agent_counter += 1
# hook_input 里有什么字段？取决于 SDK 版本
# 第一次不确定的话，先用 debug_hook 打印出来看看
# 这里假设 SubagentStart 事件会传 session_id 和 prompt
session_id = hook_input.get("session_id", tool_use_id)
prompt = hook_input.get("prompt", "未知任务")
# 给这个子Agent取一个好认的编号
agent_id = f"AGENT-{agent_counter}"
# 登记这个子Agent
tracker.sessions[session_id] = {
"id": agent_id,
"task": prompt[:50], # 只取前50个字符，太长了不好看
"tool_calls": [], # 它将来用了哪些工具，都记在这
}
tracker._current_parent_id = session_id
print(f"🚀 Spawning {agent_id}: {prompt[:50]}")
return {}
async def on_subagent_stop(hook_input, tool_use_id, context):
"""当一个子Agent干完活回来时触发"""
session_id = hook_input.get("session_id", tool_use_id)
if session_id in tracker.sessions:
session = tracker.sessions[session_id]
count = len(session["tool_calls"])
print(f"🏁 {session['id']} 完成！共调用了 {count} 次工具")
return {}
# ============ 第三步：把所有 Hook 注册到 ClaudeAgentOptions ============
async def main():
options = ClaudeAgentOptions(
system_prompt="你是研究团队的组长，负责安排子Agent完成研究任务。",
allowed_tools=["Task"],
permission_mode="bypassPermissions",
max_turns=10,
# 👇 这里是关键：把追踪器的方法挂到 hooks 上
hooks={
# 工具调用前 → 追踪器记录"谁在用什么工具"
"PreToolUse": [
HookMatcher(
matcher=None, # 匹配所有工具
hooks=[tracker.pre_tool_use_hook]
# ^^^^^^^^^^^^^^^^^^^^^^^^^^
# 注意：这里传的是实例方法，不是普通函数
# 所以 tracker 对象的 self.sessions 能跨调用保持数据
),
],
# 工具调用后 → 追踪器记录结果
"PostToolUse": [
HookMatcher(
matcher=None,
hooks=[tracker.post_tool_use_hook]
),
],
# 子Agent启动时 → 登记新员工
"SubagentStart": [
HookMatcher(
matcher=None,
hooks=[on_subagent_start]
),
],
# 子Agent结束时 → 打印完成报告
"SubagentStop": [
HookMatcher(
matcher=None,
hooks=[on_subagent_stop]
),
],
}
)
async with ClaudeSDKClient(options=options) as client:
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"\n💬 {block.text}\n")
# ============ 最后：打印完整的追踪报告 ============
print("\n" + "=" * 50)
print("📊 完整追踪报告：")
print("=" * 50)
for session_id, session in tracker.sessions.items():
print(f"\n👤 {session['id']}（任务：{session['task']}）")
for call in session["tool_calls"]:
print(f" └── {call['tool']}")
anyio.run(main)
执行起来是什么效果？
用一张时间线来看整个过程：
你说："研究一下电动汽车电池技术"
│
▼
┌─────────────── 组长 Agent ───────────────┐
│ 组长决定派 3 个子Agent │
│ │
│ ① 调用 Task 工具 → 派出研究员1 │
│ ┌──── SubagentStart Hook 触发 ────┐ │
│ │ 🚀 Spawning RESEARCHER-1 │ │
│ │ tracker 登记： │ │
│ │ sessions["xxx"] = { │ │
│ │ "id": "RESEARCHER-1", │ │
│ │ "tool_calls": [] │ │
│ │ } │ │
│ └────────────────────────────────┘ │
│ │
│ 研究员1开始干活： │
│ WebSearch("电池技术") │
│ ┌──── PreToolUse Hook 触发 ──────┐ │
│ │ [RESEARCHER-1] → WebSearch │ │
│ │ tool_calls 里多了一条记录 │ │
│ └────────────────────────────────┘ │
│ │
│ WebSearch("固态电池 2025") │
│ ┌──── PreToolUse Hook 触发 ──────┐ │
│ │ [RESEARCHER-1] → WebSearch │ │
│ └────────────────────────────────┘ │
│ │
│ Write("battery_research.md") │
│ ┌──── PreToolUse Hook 触发 ──────┐ │
│ │ [RESEARCHER-1] → Write │ │
│ └────────────────────────────────┘ │
│ │
│ ┌──── SubagentStop Hook 触发 ────┐ │
│ │ 🏁 RESEARCHER-1 完成！ │ │
│ │ 共调用了 3 次工具 │ │
│ └────────────────────────────────┘ │
│ │
│ ② 调用 Task 工具 → 派出分析师 │
│ ... 同样的流程 ... │
│ │
│ ③ 调用 Task 工具 → 派出报告员 │
│ ... 同样的流程 ... │
└───────────────────────────────────────────┘
│
▼
最终控制台输出：
🚀 Spawning RESEARCHER-1: battery technology research
[RESEARCHER-1] → WebSearch
[RESEARCHER-1] → WebSearch
[RESEARCHER-1] → Write
🏁 RESEARCHER-1 完成！共调用了 3 次工具
🚀 Spawning DATA-ANALYST-1: generate visualizations
[DATA-ANALYST-1] → Glob
[DATA-ANALYST-1] → Read
[DATA-ANALYST-1] → Bash
[DATA-ANALYST-1] → Write
🏁 DATA-ANALYST-1 完成！共调用了 4 次工具
💬 研究报告已完成...
==================================================
📊 完整追踪报告：
==================================================
👤 RESEARCHER-1（任务：battery technology research）
└── WebSearch
└── WebSearch
└── Write
👤 DATA-ANALYST-1（任务：generate visualizations）
└── Glob
└── Read
└── Bash
└── Write
关键理解：数据是怎么串起来的？
很多人会困惑：Claude 派了好几个子Agent，Hook 怎么知道"这次 WebSearch 是研究员1的，不是研究员2的"？
秘密在于 tool_use_id 这个参数：
async def pre_tool_use_hook(self, hook_input, tool_use_id, context):
# ^^^^^^^^^^^
# 每次工具调用都有唯一ID
整个流程就像公司的"工牌系统"：
1. 子Agent被派出时（SubagentStart），拿到一个工牌号（tool_use_id）
→ tracker 用这个工牌号建了一条员工档案
2. 子Agent每次用工具时（PreToolUse），出示工牌号
→ tracker 根据工牌号找到对应的员工档案
→ 在档案里添加一条"用了什么工具"的记录
3. 子Agent干完活了（SubagentStop），交回工牌号
→ tracker 根据工牌号找到档案，打印完成报告
这就是为什么 tracker 要用一个 类 来写，而不是一个普通函数——因为它需要用 self.sessions 在多次 Hook 调用之间保持记忆。如果用普通函数，每次调用都是"失忆"的，无法把同一个子Agent的多次工具调用串起来。
所有 Hook 事件类型
SDK 支持的 Hook 事件不只有工具调用前后：
| 事件 | 什么时候触发 | 常见用途 |
|---|---|---|
PreToolUse |
工具调用前 | 安全检查、日志、拦截 |
PostToolUse |
工具调用后（成功时） | 记录结果、统计 |
PostToolUseFailure |
工具调用失败时 | 错误处理 |
UserPromptSubmit |
用户发送消息时 | 过滤输入 |
Stop |
Agent 停止时 | 清理资源 |
SubagentStart |
子Agent启动时 | 追踪子Agent |
SubagentStop |
子Agent结束时 | 统计耗时 |
Notification |
系统通知 | 进度报告 |
研究平台主要用 PreToolUse 和 PostToolUse，已经够用了。
本课小结
- Hook = 监控摄像头，插在 AI 工作流程中
- PreToolUse 在工具调用前触发，可以放行或拦截
- PostToolUse 在工具调用后触发，用来记录结果
- HookMatcher 决定匹配哪些工具
- 返回空字典 = 放行，返回 deny 决策 = 拦截
- 研究平台用 Hook 追踪每个子Agent的所有活动
课后练习
- 写一个 Hook，统计整个对话中总共调用了多少次工具
- 写一个 Hook，限制 WebSearch 最多调用 3 次（超过就拦截）
- 写一个 PostToolUse Hook，把每次工具调用的结果保存到一个 JSON 文件