Documentation Index
Fetch the complete documentation index at: https://docs.qoder.com/llms.txt
Use this file to discover all available pages before exploring further.
Hooks 允许你在 AI 会话的关键生命周期节点注入自定义逻辑,支持审计日志、安全控制、上下文注入和动态行为修改。
事件概览
| 事件 | 触发时机 | 可控行为 |
|---|
PreToolUse | 工具调用前 | 拦截 / 放行 / 修改输入 |
PostToolUse | 工具执行成功后 | 审计 / 注入上下文 / 覆盖输出 |
PostToolUseFailure | 工具执行失败后 | 错误处理 / 日志记录 |
UserPromptSubmit | 用户 prompt 发送前 | 注入上下文 / 拦截 |
SessionStart | 会话开始时 | 初始化 / 注入上下文 |
SessionEnd | 会话结束时 | 清理 / 日志记录 |
Stop | AI 停止生成时 | 阻止停止,强制继续 |
SubagentStart | 子 Agent 启动时 | 观察 / 日志记录 |
SubagentStop | 子 Agent 停止时 | 观察 / 日志记录 |
PreCompact | 上下文压缩前 | 观察 / 日志记录 |
PostCompact | 上下文压缩后 | 观察 / 日志记录 |
CwdChanged | 工作目录变更时 | 观察 / 日志记录 |
InstructionsLoaded | 指令文件加载时 | 观察 / 日志记录 |
FileChanged | 文件创建/修改/删除时 | 观察 / 日志记录 |
PermissionRequest | 权限申请时 | 自动批准 / 拒绝权限请求 |
完整事件类型定义见 Hooks Reference。
在 QoderAgentOptions.hooks 中配置 hooks:
from qoder_agent_sdk import query, QoderAgentOptions, HookMatcher
async for msg in query(
prompt="perform task",
options=QoderAgentOptions(
hooks={
"PreToolUse": [HookMatcher(matcher="Bash", hooks=[my_hook])],
"PostToolUse": [HookMatcher(hooks=[audit_hook])],
"SessionEnd": [HookMatcher(hooks=[log_hook])],
},
),
):
... # process messages
Matcher
matcher 字段为正则表达式,只有工具名称匹配时 hook 才会触发:
hooks={
"PreToolUse": [
HookMatcher(matcher="Bash", hooks=[bash_audit]), # 仅 Bash
HookMatcher(matcher="File.*|Write|Edit", hooks=[file_audit]), # 文件操作
HookMatcher(hooks=[general_log]), # 所有工具(无 matcher)
],
}
回调函数
每个 hook 回调接收事件输入、工具调用 ID 和上下文:
HookCallback = Callable[
[HookInput, str | None, HookContext],
Awaitable[HookJSONOutput],
]
所有事件共享通用字段:hook_event_name(事件类型)、session_id(会话 ID)、transcript_path(记录文件路径)、cwd(工作目录)。每个事件还有专属字段,如 PreToolUse 的 tool_name 和 tool_input。
完整输入类型定义见 Hooks Reference。
回调返回一个 dict,通过以下字段控制行为:
continue_: False — 终止会话(序列化为 JSON "continue")
decision: "block" + reason — 阻止工具执行或阻止 AI 停止
hookSpecificOutput — 事件专属输出,如修改工具输入(updatedInput)、覆盖工具输出(updatedToolOutput)、注入上下文(additionalContext)
完整输出类型定义见 Hooks Reference。
拦截危险的 shell 命令:
async def security_hook(inp: HookInput, tid: str | None, ctx: HookContext) -> HookJSONOutput:
if inp.get("tool_name") == "Bash":
cmd = (inp.get("tool_input") or {}).get("command", "")
if "rm -rf" in cmd:
return {
"hookSpecificOutput": {
"hookEventName": "PreToolUse",
"permissionDecision": "deny",
"permissionDecisionReason": "Destructive delete operations are not allowed",
},
}
return {}
脱敏敏感信息(PostToolUse)
覆盖工具输出,替换 AK/Token 等敏感信息:
import re
async def secret_redact_hook(inp: HookInput, tid: str | None, ctx: HookContext) -> HookJSONOutput:
if inp.get("hook_event_name") != "PostToolUse":
return {}
response = inp.get("tool_response", "")
content = response if isinstance(response, str) else json.dumps(response)
redacted = re.sub(r"(?:LTAI|AKID)[A-Za-z0-9]{16,}", "<REDACTED_AK>", content)
redacted = re.sub(r"Bearer\s+[A-Za-z0-9\-._~+/]+=*", "Bearer <REDACTED>", redacted)
if redacted == content:
return {}
return {
"hookSpecificOutput": {
"hookEventName": "PostToolUse",
"updatedToolOutput": redacted,
},
}
裁剪过长输出(PostToolUse)
截断超长的 Bash 输出,保留头尾:
async def bash_summarize_hook(inp: HookInput, tid: str | None, ctx: HookContext) -> HookJSONOutput:
if inp.get("hook_event_name") != "PostToolUse":
return {}
if inp.get("tool_name") != "Bash":
return {}
content = str(inp.get("tool_response") or "")
THRESHOLD = 50 * 1024
if len(content) <= THRESHOLD:
return {}
head = content[:8 * 1024]
tail = content[-4 * 1024:]
omitted = len(content) - len(head) - len(tail)
return {
"hookSpecificOutput": {
"hookEventName": "PostToolUse",
"updatedToolOutput": f"{head}\n\n[... OMITTED {omitted} chars ...]\n\n{tail}",
},
}
强制继续(Stop)
阻止 AI 在任务未完成时停止:
async def keep_going(inp: HookInput, tid: str | None, ctx: HookContext) -> HookJSONOutput:
if inp.get("hook_event_name") != "Stop":
return {}
if not is_task_complete():
return {"decision": "block", "reason": "Please continue completing the remaining tasks"}
return {}
自动批准权限(PermissionRequest)
自动放行 Read 工具的权限请求:
async def auto_approve_read(inp: HookInput, tid: str | None, ctx: HookContext) -> HookJSONOutput:
if inp.get("hook_event_name") != "PermissionRequest":
return {}
if inp.get("tool_name") == "Read":
return {
"hookSpecificOutput": {
"hookEventName": "PermissionRequest",
"decision": {"behavior": "allow"},
},
}
return {}
完整权限模型请参见权限文档。
审计与安全控制(综合)
组合审计日志和安全拦截:
import json
import logging
import re
from qoder_agent_sdk import query, QoderAgentOptions, HookMatcher
from qoder_agent_sdk.types import HookInput, HookContext, HookJSONOutput
async def security_hook(inp: HookInput, tid: str | None, ctx: HookContext) -> HookJSONOutput:
if inp.get("hook_event_name") != "PreToolUse":
return {}
# Audit log
logging.info(json.dumps({
"event": "tool_call",
"tool": inp.get("tool_name"),
"input": inp.get("tool_input"),
}))
# Security check: block curl to external domains
if inp.get("tool_name") == "Bash":
cmd = str((inp.get("tool_input") or {}).get("command", ""))
if re.search(r"curl\s+https?://(?!localhost)", cmd):
return {
"hookSpecificOutput": {
"hookEventName": "PreToolUse",
"permissionDecision": "deny",
"permissionDecisionReason": "HTTP requests to external domains are not allowed",
},
}
return {}
async def main():
async for msg in query(
prompt="run deployment",
options=QoderAgentOptions(
hooks={
"PreToolUse": [HookMatcher(hooks=[security_hook])],
},
),
):
pass # process messages
注意事项
- Hook 回调应尽快返回,避免阻塞 AI 执行。
matcher 使用 Python 正则语法(re 模块),匹配 tool_name 字段。
continue_: False 可终止会话——仅对 PreToolUse、PostToolUse、PostToolUseFailure、UserPromptSubmit、Stop、SubagentStop 事件有效,观察类事件(如 SessionEnd、CwdChanged)会忽略此字段。
- 当多个 hook 返回冲突的
decision 值时,"deny" / "block" 优先(最严格的规则生效)。
- 当多个 hook 都设置了
updatedToolOutput 时,最后一个非空值生效。如需链式执行多个转换(如先脱敏再裁剪),请在单个回调内按顺序执行。
- Python SDK 使用尾部下划线字段名(
continue_)以避免与 Python 关键字冲突。SDK 在序列化时会自动转换为线路协议名称(continue)。