SDK Hook 系统

版本要求：本文档针对 CodeBuddy Agent SDK v0.1.0 及以上版本。

本文档介绍如何在 SDK 中使用 Hook 系统，在工具执行前后插入自定义逻辑。

概述

Hook 允许你在 CodeBuddy 的会话生命周期内插入自定义逻辑，实现：

工具调用前的校验和拦截
工具执行后的日志记录
用户提交内容的审查
会话开始/结束时的初始化和清理

支持的事件

事件	触发时机
`PreToolUse`	工具执行前
`PostToolUse`	工具执行成功后
`UserPromptSubmit`	用户提交消息时
`Stop`	主 Agent 响应结束时
`SubagentStop`	子 Agent 结束时
`PreCompact`	上下文压缩前
`unstable_Checkpoint`	文件修改后自动创建检查点时

Hook 配置

通过 hooks 选项配置 Hook。每个事件可以有多个 matcher，每个 matcher 可以有多个 hook 回调。

基本结构

TypeScriptPython

typescript

import { query } from '@tencent-ai/agent-sdk';

const q = query({
  prompt: '帮我分析代码',
  options: {
    model: 'deepseek-v3.1',
    hooks: {
      PreToolUse: [
        {
          matcher: 'Bash',  // 只匹配 Bash 工具
          hooks: [
            async (input, toolUseId, ctx) => {
              console.log('即将执行：', input);
              return { continue: true };
            }
          ],
          timeout：5000  // 超时时间（毫秒）
        }
      ]
    }
  }
});

python

from codebuddy_agent_sdk import query, CodeBuddyAgentOptions, HookMatcher

async def pre_tool_hook(input_data, tool_use_id, context):
    print(f"即将执行： {input_data}")
    return {"continue_": True}

options = CodeBuddyAgentOptions(
    model="deepseek-v3.1",
    hooks={
        "PreToolUse": [
            HookMatcher(
                matcher="Bash",  # 只匹配 Bash 工具
                hooks=[pre_tool_hook],
                timeout=5.0  # 超时时间（秒）
            )
        ]
    }
)

async for msg in query(prompt="帮我分析代码", options=options):
    print(msg)

HookMatcher 结构

字段	类型	说明
`matcher`	`string`	匹配模式,支持正则表达式。`*` 或空字符串匹配所有
`hooks`	`HookCallback[]`	回调函数数组
`timeout`	`number`	超时时间（TypeScript 毫秒，Python 秒）

Matcher 模式

精确匹配："Bash" 只匹配 Bash 工具
正则匹配："Edit|Write" 匹配 Edit 或 Write
通配符："*" 或 "" 匹配所有工具
前缀匹配："mcp__.*" 匹配所有 MCP 工具

事件类型

PreToolUse

工具执行前触发，可以阻止执行或修改输入。

TypeScriptPython

typescript

hooks: {
  PreToolUse: [{
    matcher: 'Bash',
    hooks: [
      async (input, toolUseId, ctx) => {
        const command = input.command as string;

        // 阻止危险命令
        if (command.includes('rm -rf')) {
          return {
            decision: 'block',
            reason: '危险命令被阻止'
          };
        }

        return { continue: true };
      }
    ]
  }]
}

python

async def pre_bash_hook(input_data, tool_use_id, context):
    command = input_data.get("command", "")

    # 阻止危险命令
    if "rm -rf" in command:
        return {
            "decision": "block",
            "reason": "危险命令被阻止"
        }

    return {"continue_": True}

hooks = {
    "PreToolUse": [
        HookMatcher(matcher="Bash", hooks=[pre_bash_hook])
    ]
}

PostToolUse

工具执行成功后触发，可以添加额外上下文。

TypeScriptPython

typescript

hooks: {
  PostToolUse: [{
    matcher: 'Write|Edit',
    hooks: [
      async (input, toolUseId) => {
        console.log(`文件已修改: ${input.file_path}`);
        // 记录修改日志
        await logFileChange(input.file_path);
        return { continue: true };
      }
    ]
  }]
}

python

async def post_write_hook(input_data, tool_use_id, context):
    print(f"文件已修改： {input_data.get('file_path')}")
    # 记录修改日志
    await log_file_change(input_data.get("file_path"))
    return {"continue_": True}

hooks = {
    "PostToolUse": [
        HookMatcher(matcher="Write|Edit", hooks=[post_write_hook])
    ]
}

UserPromptSubmit

用户提交消息时触发，可以添加上下文或阻止处理。

TypeScriptPython

typescript

hooks: {
  UserPromptSubmit: [{
    hooks: [
      async (input) => {
        const prompt = input.prompt as string;

        // 敏感词检查
        if (containsSensitiveWords(prompt)) {
          return {
            decision: 'block',
            reason: '消息包含敏感内容'
          };
        }

        return { continue: true };
      }
    ]
  }]
}

python

async def prompt_check_hook(input_data, tool_use_id, context):
    prompt = input_data.get("prompt", "")

    # 敏感词检查
    if contains_sensitive_words(prompt):
        return {
            "decision": "block",
            "reason": "消息包含敏感内容"
        }

    return {"continue_": True}

hooks = {
    "UserPromptSubmit": [
        HookMatcher(hooks=[prompt_check_hook])
    ]
}

Stop / SubagentStop

Agent 响应结束时触发，可以阻止停止并要求继续。

TypeScriptPython

typescript

hooks: {
  Stop: [{
    hooks: [
      async (input) => {
        // 检查任务是否真正完成
        if (!isTaskComplete()) {
          return {
            decision: 'block',
            reason: '任务未完成，请继续'
          };
        }
        return { continue: true };
      }
    ]
  }]
}

python

async def stop_hook(input_data, tool_use_id, context):
    # 检查任务是否真正完成
    if not is_task_complete():
        return {
            "decision": "block",
            "reason": "任务未完成，请继续"
        }
    return {"continue_": True}

hooks = {
    "Stop": [HookMatcher(hooks=[stop_hook])]
}

unstable_Checkpoint（实验性）

文件修改后（Write/Edit/MultiEdit 工具执行成功）自动触发，提供文件快照和变更统计信息。

实验性 API

此 Hook 为实验性功能，API 可能在未来版本中发生变化。

TypeScriptPython

typescript

import type { CheckpointHookInput } from '@tencent-ai/agent-sdk';

hooks: {
  unstable_Checkpoint: [{
    hooks: [
      async (input) => {
        const checkpointInput = input as CheckpointHookInput;
        const checkpoint = checkpointInput.checkpoint;
        
        console.log('文件变更检查点：', {
          id: checkpoint.id,
          label: checkpoint.label,
          files: checkpoint.fileChangeStats?.files,
          additions: checkpoint.fileChangeStats?.additions,
          deletions: checkpoint.fileChangeStats?.deletions
        });
        
        // 访问文件快照
        for (const [filePath, version] of Object.entries(checkpoint.fileSnapshots)) {
          console.log(`  ${filePath} - 版本 ${version.version}`);
        }
        
        return { continue: true };
      }
    ]
  }]
}

python

async def checkpoint_hook(input_data, tool_use_id, context):
    checkpoint = input_data.get("checkpoint", {})
    
    file_change_stats = checkpoint.get("fileChangeStats", {})
    print(f"文件变更检查点：")
    print(f"  ID: {checkpoint.get('id')}")
    print(f"  Label: {checkpoint.get('label')}")
    print(f"  Files: {file_change_stats.get('files', [])}")
    print(f"  Additions: +{file_change_stats.get('additions', 0)} lines")
    print(f"  Deletions: -{file_change_stats.get('deletions', 0)} lines")
    
    # 访问文件快照
    for file_path, version in checkpoint.get("fileSnapshots", {}).items():
        print(f"  {file_path} - 版本 {version.get('version')}")
    
    return {"continue_": True}

hooks = {
    "unstable_Checkpoint": [HookMatcher(hooks=[checkpoint_hook])]
}

Checkpoint 数据结构：

id: 检查点唯一标识符
label: 人类可读标签（通常为用户提示）
createdAt: 创建时间戳
fileSnapshots: 文件路径到版本信息的映射
- filePath: 文件绝对路径
- version: 版本号
- backupFileName: 备份文件名
- backupTime: 备份时间戳
fileChangeStats: 文件变更统计
- files: 变更的文件路径列表
- additions: 新增行数
- deletions: 删除行数

Hook 输入

Hook 回调接收的输入结构因事件类型而异。

公共字段

json

{
  "session_id": "abc123",
  "cwd": "/path/to/project",
  "permission_mode": "default",
  "hook_event_name": "PreToolUse"
}

PreToolUse / PostToolUse 输入

json

{
  "tool_name": "Bash",
  "tool_input": {
    "command": "ls -la"
  }
}

UserPromptSubmit 输入

json

{
  "prompt": "帮我写一个函数"
}

Stop / SubagentStop 输入

json

{
  "stop_hook_active": false
}

unstable_Checkpoint 输入

json

{
  "checkpoint": {
    "id": "ckpt_abc123",
    "label": "帮我写一个函数",
    "createdAt": 1705920000000,
    "fileSnapshots": {
      "/path/to/file.ts": {
        "filePath": "/path/to/file.ts",
        "version": 1,
        "backupFileName": "file.ts.v1.backup",
        "backupTime": 1705920000000
      }
    },
    "fileChangeStats": {
      "files": ["/path/to/file.ts"],
      "additions": 10,
      "deletions": 2
    }
  }
}

Hook 输出

Hook 回调返回的输出控制后续行为。

基本输出字段

字段	类型	说明
`continue` / `continue_`	`boolean`	是否继续执行（默认 true）
`decision`	`'block'`	设为 `'block'` 阻止操作
`reason`	`string`	阻止原因
`stopReason`	`string`	当 `continue` 为 false 时显示的停止消息
`suppressOutput`	`boolean`	隐藏输出

PreToolUse 特殊输出

可以修改工具输入：

TypeScriptPython

typescript

return {
  continue: true,
  hookSpecificOutput: {
    hookEventName: 'PreToolUse',
    updatedInput: {
      command: `echo "安全检查通过" && ${input.command}`
    }
  }
};

python

return {
    "continue_": True,
    "hookSpecificOutput": {
        "hookEventName": "PreToolUse",
        "updatedInput": {
            "command": f'echo "安全检查通过" && {input_data["command"]}'
        }
    }
}

示例

完整示例：Bash 命令审计

TypeScriptPython

typescript

import { query } from '@tencent-ai/agent-sdk';
import * as fs from 'fs';

const logFile = '/tmp/bash-audit.log';

const q = query({
  prompt: '帮我清理临时文件',
  options: {
    model: 'deepseek-v3.1',
    hooks: {
      PreToolUse: [{
        matcher: 'Bash',
        hooks: [
          async (input, toolUseId) => {
            const command = input.command as string;
            const timestamp = new Date().toISOString();

            // 记录命令
            fs.appendFileSync(logFile, `${timestamp} [PRE] ${command}\n`);

            // 危险命令检查
            const dangerous = ['rm -rf /', 'mkfs', ':(){:|:&};:'];
            for (const d of dangerous) {
              if (command.includes(d)) {
                return {
                  decision: 'block',
                  reason: `危险命令被阻止: ${d}`
                };
              }
            }

            return { continue: true };
          }
        ]
      }],
      PostToolUse: [{
        matcher: 'Bash',
        hooks: [
          async (input, toolUseId) => {
            const command = input.command as string;
            const timestamp = new Date().toISOString();

            // 记录执行完成
            fs.appendFileSync(logFile, `${timestamp} [POST] ${command} - 完成\n`);

            return { continue: true };
          }
        ]
      }]
    }
  }
});

for await (const message of q) {
  console.log(message);
}

python

import asyncio
from datetime import datetime
from codebuddy_agent_sdk import query, CodeBuddyAgentOptions, HookMatcher

log_file = "/tmp/bash-audit.log"

async def pre_bash_hook(input_data, tool_use_id, context):
    command = input_data.get("command", "")
    timestamp = datetime.now().isoformat()

    # 记录命令
    with open(log_file, "a") as f:
        f.write(f"{timestamp} [PRE] {command}\n")

    # 危险命令检查
    dangerous = ["rm -rf /", "mkfs", ":(){:|:&};:"]
    for d in dangerous:
        if d in command:
            return {
                "decision": "block",
                "reason": f"危险命令被阻止： {d}"
            }

    return {"continue_": True}

async def post_bash_hook(input_data, tool_use_id, context):
    command = input_data.get("command", "")
    timestamp = datetime.now().isoformat()

    # 记录执行完成
    with open(log_file, "a") as f:
        f.write(f"{timestamp} [POST] {command} - 完成\n")

    return {"continue_": True}

async def main():
    options = CodeBuddyAgentOptions(
        model="deepseek-v3.1",
        hooks={
            "PreToolUse": [
                HookMatcher(matcher="Bash", hooks=[pre_bash_hook])
            ],
            "PostToolUse": [
                HookMatcher(matcher="Bash", hooks=[post_bash_hook])
            ]
        }
    )

    async for message in query(prompt="帮我清理临时文件", options=options):
        print(message)

asyncio.run(main())

示例：限制文件修改范围

TypeScriptPython

typescript

hooks: {
  PreToolUse: [{
    matcher: 'Write|Edit',
    hooks: [
      async (input) => {
        const filePath = input.file_path as string;

        // 只允许修改 src 目录
        if (!filePath.startsWith('/path/to/project/src/')) {
          return {
            decision: 'block',
            reason: `不允许修改 src 目录外的文件: ${filePath}`
          };
        }

        // 禁止修改配置文件
        if (filePath.endsWith('.env') || filePath.includes('.git/')) {
          return {
            decision: 'block',
            reason: '不允许修改敏感文件'
          };
        }

        return { continue: true };
      }
    ]
  }]
}

python

async def file_scope_hook(input_data, tool_use_id, context):
    file_path = input_data.get("file_path", "")

    # 只允许修改 src 目录
    if not file_path.startswith("/path/to/project/src/"):
        return {
            "decision": "block",
            "reason": f"不允许修改 src 目录外的文件： {file_path}"
        }

    # 禁止修改配置文件
    if file_path.endswith(".env") or ".git/" in file_path:
        return {
            "decision": "block",
            "reason": "不允许修改敏感文件"
        }

    return {"continue_": True}

hooks = {
    "PreToolUse": [
        HookMatcher(matcher="Write|Edit", hooks=[file_scope_hook])
    ]
}

示例：文件修改追踪（Checkpoint Hook）

TypeScriptPython

typescript

import { query, type CheckpointHookInput } from '@tencent-ai/agent-sdk';
import * as fs from 'fs';

const changeLog = '/tmp/file-changes.log';

const q = query({
  prompt: '重构 src/utils.ts 文件',
  options: {
    model: 'deepseek-v3.1',
    hooks: {
      unstable_Checkpoint: [{
        hooks: [
          async (input) => {
            const checkpointInput = input as CheckpointHookInput;
            const checkpoint = checkpointInput.checkpoint;
            const stats = checkpoint.fileChangeStats;
            
            if (!stats) return { continue: true };
            
            // 记录文件变更
            const timestamp = new Date().toISOString();
            const logEntry = `
[${timestamp}] Checkpoint ${checkpoint.id}
  Label: ${checkpoint.label}
  Files: ${stats.files.join(', ')}
  Changes: +${stats.additions}/-${stats.deletions}
  Snapshots: ${Object.keys(checkpoint.fileSnapshots).length} files
`;
            
            fs.appendFileSync(changeLog, logEntry);
            
            // 如果变更过大，提醒用户
            if (stats.additions + stats.deletions > 100) {
              console.warn('⚠️  大量代码变更，建议review');
            }
            
            return { continue: true };
          }
        ]
      }]
    }
  }
});

for await (const message of q) {
  console.log(message);
}

python

import asyncio
from datetime import datetime
from codebuddy_agent_sdk import query, CodeBuddyAgentOptions, HookMatcher

change_log = "/tmp/file-changes.log"

async def checkpoint_tracker(input_data, tool_use_id, context):
    checkpoint = input_data.get("checkpoint", {})
    stats = checkpoint.get("fileChangeStats")
    
    if not stats:
        return {"continue_": True}
    
    # 记录文件变更
    timestamp = datetime.now().isoformat()
    log_entry = f"""
[{timestamp}] Checkpoint {checkpoint.get('id')}
  Label: {checkpoint.get('label')}
  Files: {', '.join(stats.get('files', []))}
  Changes: +{stats.get('additions', 0)}/-{stats.get('deletions', 0)}
  Snapshots: {len(checkpoint.get('fileSnapshots', {}))} files
"""
    
    with open(change_log, "a") as f:
        f.write(log_entry)
    
    # 如果变更过大，提醒用户
    total_changes = stats.get("additions", 0) + stats.get("deletions", 0)
    if total_changes > 100:
        print("⚠️  大量代码变更，建议 review")
    
    return {"continue_": True}

async def main():
    options = CodeBuddyAgentOptions(
        model="deepseek-v3.1",
        hooks={
            "unstable_Checkpoint": [
                HookMatcher(hooks=[checkpoint_tracker])
            ]
        }
    )
    
    async for message in query(prompt="重构 src/utils.ts 文件", options=options):
        print(message)

asyncio.run(main())

SDK Hook 系统 ​

概述 ​

支持的事件 ​

Hook 配置 ​

基本结构 ​

HookMatcher 结构 ​

Matcher 模式 ​

事件类型 ​

PreToolUse ​

PostToolUse ​

UserPromptSubmit ​

Stop / SubagentStop ​

unstable_Checkpoint（实验性） ​

Hook 输入 ​

公共字段 ​

PreToolUse / PostToolUse 输入 ​

UserPromptSubmit 输入 ​

Stop / SubagentStop 输入 ​

unstable_Checkpoint 输入 ​

Hook 输出 ​

基本输出字段 ​

PreToolUse 特殊输出 ​

示例 ​

完整示例：Bash 命令审计 ​

示例：限制文件修改范围 ​

示例：文件修改追踪（Checkpoint Hook） ​

相关文档 ​

SDK Hook 系统

概述

支持的事件

Hook 配置

基本结构

HookMatcher 结构

Matcher 模式

事件类型

PreToolUse

PostToolUse

UserPromptSubmit

Stop / SubagentStop

unstable_Checkpoint（实验性）

Hook 输入

公共字段

PreToolUse / PostToolUse 输入

UserPromptSubmit 输入

Stop / SubagentStop 输入

unstable_Checkpoint 输入

Hook 输出

基本输出字段

PreToolUse 特殊输出

示例

完整示例：Bash 命令审计

示例：限制文件修改范围

示例：文件修改追踪（Checkpoint Hook）

相关文档