手搓Claude Code-第四章 hooks
手搓Claude Code-第四章 hooks
写在前面
第四章讲述了钩子，让agent_loop变得更整洁的操作。完整代码见
https://github.com/shareAI-lab/learn-claude-code/blob/main/s04_hooks/code.py
我们的任务是：
1，了解hooks，清楚hooks的重要性
2，将s03的代码一步步重构为hooks
3，感受引入hooks之后的模型交互过程
还记得第三章末尾我遇到的问题吗？在真实环境中往往会编写一些日志函数去实时监控模型的工作状态，那监控有很多种，有的在调用结束时，有的在调用之前，如果我们想把这些都加进去，那整个agent_loop会显得十分臃肿复杂，这十分不规范。因此，我们本着重构的思想，引入hooks，使行文变得简洁可拓展，当加入一些类似日志函数的辅助功能时，变得容易。重点感受Claude Code系统的规范和拓展性。
一、什么是Hooks（钩子）
先欣赏灵魂画作。。。
将循环想象成一个衣架，原本塞在循环里的各种功能（比如权限检查、日志、自动 git）拆出来，挂到固定的 “钩子” 上，让循环本身保持干净、稳定。想加东西就挂上去，不用拆衣架本身。
这样避免每次加新功能都要改循环，改多了容易错，难维护。
其实在第三章中，我们写的check_permission，就是把三个门组装在一起的函数，就是一个hook，不过还应注意，所有的hook返回值类型都应该还是一样的。此外，我们还应该学习如何管理它们。
二、将s03的新增代码重构，逻辑不变。
第一步，我们先搭建一个HOOK的基础框架，这是整个机制的核心。不管这些钩子具体是做什么，都要注册到固定的事件点上，同时需要一个通用的触发器，循环只负责在对应节点触发钩子函数。
先写注册表，我们暂时把事件节点简单分为以下四个。注册表是一个字典，事件节点作为键，对应一个列表，用来存放钩子函数。
# 注册表
HOOKS = {
"UserPromptSubmit": [], # 用户输入后，模型思考前
"PreToolUse": [], # 执行工具前
"PostToolUse": [], # 执行工具后
"Stop": [], # 得到一次任务的最终结果后，循环退出前。
}
通用的注册函数，用来注册钩子，把自定义的钩子函数绑定到对应的事件上。callback就是钩子函数的名称。
def register_hook(event: str, callback):
HOOKS[event].append(callback)
通用的触发器，用于在指定的事件点，执行所有绑定的钩子函数。执行一遍对应事件点下所有的钩子函数，一旦有结果就返回。都没有结果则返回None，这提醒我们后面钩子函数的返回值类型，要么是一个字符串，要么是None。
def trigger_hooks(event: str, *args):
for callback in HOOKS[event]:
result = callback(*args)
if result is not None:
return result
return None
第二步，修改一个个功能为一个个钩子。先趁热来修改s03更新的三个门，直接贴在下面。
def permission_hook(block):
"""
原来的s03中，又是门1，2，3太乱了，这里统一给他组装成hook，规定标准的返回类型
"""
if block.name == "bash":
# 第一道门
for pattern in DENY_LIST:
if pattern in block.input.get("command", ""):
print(f"\n\033[31m⛔(门1) Blocked: '{pattern}'\033[0m")
return "Permission denied by deny list"
# 当调用bash的时候, 第二和三个门
for kw in DESTRUCTIVE:
if kw in block.input.get("command"):
print(f"\n\033[33m⚠ Potentially destructive command\033[0m")
print(f" Tool: {block.name}({block.input})")
choice = input(" Allow? [y/N]").strip().lower()
if choice not in ("y", "yes"):
return "Permission denied by user"
# 当调用工具的时候，第二和第三个门
if block.name in ("write_file", "edit_file"):
path = block.input.get("path", "")
if not (WORKDIR / path).resolve().is_relative_to(WORKDIR):
print(f"\n\033[33m⚠ Writing outside workspace\033[0m")
print(f" Tool: {block.name}({block.input})")
if choice not in ("y", "yes"):
return "Permission denied by user"
return None
其实就是重构了代码结构，所有的逻辑都是不变的。
【原 s03 权限逻辑】
函数开始 → 判断是不是bash工具
├─ 是bash：先跑黑名单拦截
│ └─ 命中黑名单 → 红色打印禁止理由，无人工确认，return False 直接阻断
└─ 不管是不是bash，都进入通用权限规则校验（第二道门）
├─ 命中风险规则 → 弹出窗口让你手动确认
│ ├─ 选deny → return False，拒绝执行
│ └─ 选allow → 放行，继续执行
└─ 没命中任何规则 → return True，正常放行
【现 s04 permission_hook 逻辑（Hook 改造后）】
函数开始 → 判断是不是bash工具
├─ 是bash：先跑黑名单(门1)拦截
│ └─ 命中黑名单 → 打印红色提示 + return 描述字符串 → 直接阻断
│ └─ 未命中黑名单 → 检测高危命令(风险规则)
│ ├─ 命中高危命令 → 弹窗手动确认
│ │ ├─ 选择拒绝 → return 描述字符串 → 阻断
│ │ └─ 选择允许 → 继续往下走
│ └─ 无高危命令 → 继续往下走
└─ 不管是不是bash，都进入通用文件权限校验
├─ 工具为 write_file/edit_file 且路径越界 → 读取用户选择
│ ├─ 选择拒绝 → return 描述字符串 → 阻断
│ └─ 选择允许 → 继续往下走
└─ 工具非文件操作 / 路径合法 → 无操作
└─ 所有规则全部通过 → return None → 正常放行
第三步，来实现一些日志函数，主要目的是在控制台打印一些参数。实现顺序与原项目略有不同，笔者认为这样更容易理解，与注册表的事件节点顺序是一样的。
context_inject_hook。在用户提交任务后，打印当前的工作目录。
def context_inject_hook(query: str):
print(f"\033[90m[HOOK] UserPromptSubmit: working in {WORKDIR}\033[0m")
return None
log_hook。在模型准备调工具前，打印block里的部分内容。
def log_hook(block):
args_preview = str(list(block.input.values())[:2])[:60] # 作为dem，仅打印前两个就行。
print(f"\033[90m[HOOK] {block.name}({args_preview})\033[0m")
return None
large_output_hook。在模型调工具之后，打印输出的结果。
def large_output_hook(block, output):
if len(str(output)) > 100000:
print(f"\033[33m[HOOK] ⚠ Large output from {block.name}: {len(str(output))} chars\033[0m")
return None
最后一个，summary_hook。
def summary_hook(message: list):
"""
这是一个后后置钩子，在一轮循环结束之后，总结调用了多少工具，这里把他归为了停止调用stop
"""
"""
isinstance()是在判断变量是不是属于某个数据类型，返回布尔值
"""
tool_count = sum(1 for m in message
for b in (m.get("content") if isinstance(m.get("context"), list) else [])
if isinstance(b, dict) and b.get("type") == "tool_result")
print(f"\033[90m[HOOK] Stop: session used {tool_count} tool calls\033[0m")
return None
中间这个生成器等价于
tool_count = 0
# 遍历每条对话消息
for m in messages:
# 取出content，不是列表就用空列表
content = m.get("content")
if not isinstance(content, list):
blocks = []
else:
blocks = content
# 遍历消息内每一个内容块
for b in blocks:
# 判断是否是tool_result块
if isinstance(b, dict) and b.get("type") == "tool_result":
tool_count += 1
然后将它们都注册一遍。
register_hook("UserPromptSubmit", context_inject_hook)
register_hook("PreToolUse", permission_hook)
register_hook("PreToolUse", log_hook)
register_hook("PostToolUse", large_output_hook)
register_hook("Stop", summary_hook)
ok，现在我们已经完成了所有Hooks，稍微暂停一下，来捋捋逻辑。以PreToolUse这个事件点为例。
这个事件下挂了两个hook，而列表是有序的。当这个事件点被触发后，观察触发器，只要前面的hook有返回值，就会停止了。所以其实hook注册的越早，在事件点的列表位置越靠前，检查的频率也就越高。只要有一个hook出现问题，就推倒重新再来。如果所有返回值都是None，才会通过。
现在，我们要做的就是在循环体对应位置去触发事件，按照时间顺序来。下面这些代码可能与原项目略有差别，原项目做了进一步的精简，逻辑都是一样的。
首先在用户提交后去触发，也就是一个任务开始前，那应该在while里面，用户输入之后就可以，见注释。
# ── Entry point ──────────────────────────────────────────
if __name__ == "__main__":
print("s04: Hooks — extension logic on hooks, loop stays clean")
print("Type a question, press Enter. Type q to quit.\n")
history = []
while True:
try:
query = input("\033[36ms01 >> \033[0m")
except (EOFError, KeyboardInterrupt):
break
if query.strip().lower() in ("q", "exit", ""):
break
# 在添加之前去触发
trigger_hooks("UserPromptSubmit", query)
history.append({"role": "user", "content": query})
agent_loop(history)
略。。。
之后去触发PreToolUse，那应该在确定要调用工具的时候，也就是agent_loop中的block.name==“tool_use”，并且如果返回值存在的话，我们得加到result中去。接下来是工具调用后，触发PostToolUse事件。
# 上略。。。
if block.type == "tool_use":
# 触发PreToolUse
blocked = trigger_hooks("PreToolUse", block)
# 添加
if blocked:
results.append({"type": "tool_result", "tool_use_id": block.id,
"content": str(blocked)})
continue
handler = TOOL_HANDLERS.get(block.name)
output = handler(**block.input) if handler else f"Unknown: {block.name}"
# 触发PostToolUse
trigger_hooks("PostToolUse", block, output)
results.append({
"type": "tool_result",
"tool_use_id": block.id,
"content": output,
})
# 下略
最后触发stop事件，应该是response.stop_reason != “tool_use”，也就是在执行bash的时候。去触发stop，要是存在返回值就说明有些门检测没通过。添加到message中去，而block循环中要添加到result中去是因为模型有可能一次性调用多个工具，会产生blocks，result最后也是要加到message中去，作为一次任务的完整信息。
# 将模型的回答传到message中，
messages.append({"role": "assistant", "content": response.content})
# 触发stop
if response.stop_reason != "tool_use":
force = trigger_hooks("Stop", messages)
if force:
messages.append({"role": "user", "content": force})
continue
return
至此，全部完成。我们来做几个测试。先来让它读一个文件。可以看到模型很出色的完成了任务。
s01 >> Read the file README.md
[HOOK] UserPromptSubmit: working in /Users/bx/Documents/coding/learn_cladudecode
观察content的内容：[ThinkingBlock(signature='pmvrgfvxxj', thinking='The user wants me to read the README.md file.', type='thinking'), ToolUseBlock(id='019ed0ccb30edcb7d82fa2d7c703c8d6', caller=None, input={'path': '/Users/bx/Documents/coding/learn_cladudecode/README.md'}, name='read_file', type='tool_use')]
[HOOK] read_file(['/Users/bx/Documents/coding/learn_cladudecode/README.md'])
观察content的内容：[ThinkingBlock(signature='yipwmhnlnm', thinking="The file doesn't exist. Let me check what files are in the directory.", type='thinking'), TextBlock(citations=None, text="The file `README.md` doesn't exist in the project root. Let me check what files are present in the directory:", type='text'), ToolUseBlock(id='019ed0ccc3f1cb764598162a488e5e4b', caller=None, input={'command': 'ls -la /Users/bx/Documents/coding/learn_cladudecode/'}, name='bash', type='tool_use')]
[HOOK] bash(['ls -la /Users/bx/Documents/coding/learn_cladudecode/'])
观察content的内容：[ThinkingBlock(signature='zhlhaxipsc', thinking="There is no README.md file in the project root. There are subdirectories for different sections. Let me check if there's a README inside any of those subdirectories.", type='thinking'), TextBlock(citations=None, text="There's no `README.md` at the project root. The directory contains subdirectories (`s01_agentLoop`, `s02_tooluse`, `s03_permission`, `s04_hooks`). Let me check if there's a README in any of them:", type='text'), ToolUseBlock(id='019ed0ccd3a1255108a019512602ded5', caller=None, input={'pattern': '**/README.md'}, name='glob', type='tool_use')]
[HOOK] glob(['**/README.md'])
观察content的内容：[TextBlock(citations=None, text="Found one! It's inside `s04_hooks`. Let me read it:", type='text'), ToolUseBlock(id='019ed0ccdc2202218c4f5fbd7b3e5385', caller=None, input={'path': '/Users/bx/Documents/coding/learn_cladudecode/s04_hooks/README.md'}, name='read_file', type='tool_use')]
[HOOK] read_file(['/Users/bx/Documents/coding/learn_cladudecode/s04_hooks/REA)
[HOOK] Stop: session used 0 tool calls
Here's the content of the only `README.md` in the project, located at `s04_hooks/README.md`:
---
This is the **s04: Hooks** section of a learning project about Claude Code internals. The key points are:
### Problem
In s03, adding 略。。。
再来尝试一个删除任务。仅多次测试，几乎没有出现上次那种情况了。看来整齐的工具调用流程不仅能让用户操作变得流畅，也能使模型调用工具操作变得流畅。
s01 >> Delete "test1.txt" file in /tmp
[HOOK] UserPromptSubmit: working in /Users/bx/Documents/coding/learn_cladudecode
观察content的内容：[ThinkingBlock(signature='zjitjkvkcu', thinking='The user is asking me to delete a file "test1.txt" in /tmp. Let me just do it.', type='thinking'), ToolUseBlock(id='019ede1d2e22290a130c3850bb13e247', caller=None, input={'command': 'rm /tmp/test1.txt'}, name='bash', type='tool_use')]
⚠ Potentially destructive command
Tool: bash({'command': 'rm /tmp/test1.txt'})
Allow? [y/N]n
[HOOK] Stop: session used 0 tool calls
The deletion was blocked — the permission hook denied the operation. This is likely because the project's hook system (as seen in s04) has a deny list that blocks `rm` commands targeting `/tmp`.
Would you like me to try a different approach, or would you like to adjust the permission settings?
总结
引入了hook使整个系统增加了更多的可扩展性。接下来我们思考另一个问题：如何节省模型的输出token？如何让模型有条理的工作？而不是让模型一直撞墙才知道回头。
欢迎加入DeepSeek 技术社区。在这里，你可以找到志同道合的朋友，共同探索AI技术的奥秘。
更多推荐
7
0- 0
已为社区贡献2条内容
所有评论(0)