Skip to content

SDK Hook 系统

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

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

概述

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

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

支持的事件

事件触发时机
PreToolUse工具执行前
PostToolUse工具执行成功后
UserPromptSubmit用户提交消息时
Stop主 Agent 响应结束时
SubagentStop子 Agent 结束时
PreCompact上下文压缩前

Hook 配置

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

基本结构

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 结构

字段类型说明
matcherstring匹配模式,支持正则表达式。* 或空字符串匹配所有
hooksHookCallback[]回调函数数组
timeoutnumber超时时间(TypeScript 毫秒,Python 秒)

Matcher 模式

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

事件类型

PreToolUse

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

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

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

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

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

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 响应结束时触发,可以阻止停止并要求继续。

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])]
}

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
}

Hook 输出

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

基本输出字段

字段类型说明
continue / continue_boolean是否继续执行(默认 true)
decision'block'设为 'block' 阻止操作
reasonstring阻止原因
stopReasonstringcontinue 为 false 时显示的停止消息
suppressOutputboolean隐藏输出

PreToolUse 特殊输出

可以修改工具输入:

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 命令审计

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())

示例:限制文件修改范围

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])
    ]
}

相关文档