使用 hooks 拦截和控制代理行为

在代理执行的关键点使用 hooks 拦截和自定义代理行为

Hooks 是回调函数，用于响应代理事件（如工具被调用、会话启动或执行停止）运行您的代码。使用 hooks，您可以：

• 阻止危险操作在执行前进行，如破坏性 shell 命令或未授权的文件访问
• 记录和审计每个工具调用，用于合规性、调试或分析
• 转换输入和输出以清理数据、注入凭证或重定向文件路径
• 要求人工批准敏感操作，如数据库写入或 API 调用
• 跟踪会话生命周期以管理状态、清理资源或发送通知

本指南涵盖 hooks 的工作原理、如何配置它们，并提供常见模式的示例，如阻止工具、修改输入和转发通知。

Hooks 如何工作

1. 事件触发：代理执行期间发生某事，SDK 触发事件：工具即将被调用（PreToolUse）、工具返回结果（PostToolUse）、子代理启动或停止、代理空闲或执行完成。
2. SDK 收集已注册的 hooks：SDK 检查为该事件类型注册的 hooks。这包括您在 options.hooks 中传递的回调 hooks 和来自设置文件的 shell 命令 hooks，当相应的 settingSources 或 setting_sources 条目启用时。
3. 匹配器过滤哪些 hooks 运行：如果 hook 有 matcher 模式（如 "Write|Edit"），SDK 会针对事件的目标（例如工具名称）测试它。没有匹配器的 hooks 对该类型的每个事件都运行。
4. 回调函数执行：每个匹配的 hook 的回调函数接收有关正在发生的事情的输入：工具名称、其参数、会话 ID 和其他事件特定的详细信息。
5. 您的回调返回决定：执行任何操作（日志记录、API 调用、验证）后，您的回调返回一个输出对象，告诉代理该做什么：允许操作、阻止它、修改输入或将上下文注入到对话中。

以下示例将这些步骤组合在一起。它注册一个 PreToolUse hook，带有 "Write|Edit" 匹配器，因此回调仅对文件写入工具触发。触发时，回调接收工具的输入，检查文件路径是否针对 .env 文件，并返回 permissionDecision: "deny" 以阻止操作：

Python：

import asyncio
from claude_agent_sdk import (
    AssistantMessage,
    ClaudeSDKClient,
    ClaudeAgentOptions,
    HookMatcher,
    ResultMessage,
)


async def protect_env_files(input_data, tool_use_id, context):
    file_path = input_data["tool_input"].get("file_path", "")
    file_name = file_path.split("/")[-1]

    if file_name == ".env":
        return {
            "hookSpecificOutput": {
                "hookEventName": input_data["hook_event_name"],
                "permissionDecision": "deny",
                "permissionDecisionReason": "Cannot modify .env files",
            }
        }

    return {}


async def main():
    options = ClaudeAgentOptions(
        hooks={
            "PreToolUse": [HookMatcher(matcher="Write|Edit", hooks=[protect_env_files])]
        }
    )

    async with ClaudeSDKClient(options=options) as client:
        await client.query("Update the database configuration")
        async for message in client.receive_response():
            if isinstance(message, (AssistantMessage, ResultMessage)):
                print(message)


asyncio.run(main())

TypeScript：

import { query, HookCallback, PreToolUseHookInput } from "@anthropic-ai/claude-agent-sdk";

const protectEnvFiles: HookCallback = async (input, toolUseID, { signal }) => {
  const preInput = input as PreToolUseHookInput;
  const toolInput = preInput.tool_input as Record<string, unknown>;
  const filePath = toolInput?.file_path as string;
  const fileName = filePath?.split("/").pop();

  if (fileName === ".env") {
    return {
      hookSpecificOutput: {
        hookEventName: preInput.hook_event_name,
        permissionDecision: "deny",
        permissionDecisionReason: "Cannot modify .env files"
      }
    };
  }

  return {};
};

for await (const message of query({
  prompt: "Update the database configuration",
  options: {
    hooks: {
      PreToolUse: [{ matcher: "Write|Edit", hooks: [protectEnvFiles] }]
    }
  }
})) {
  if (message.type === "assistant" || message.type === "result") {
    console.log(message);
  }
}

可用的 hooks

SDK 为代理执行的不同阶段提供 hooks。某些 hooks 在两个 SDK 中都可用，而其他 hooks 仅在 TypeScript 中可用。

Hook 事件	Python SDK	TypeScript SDK	触发条件	示例用例
PreToolUse	是	是	工具调用请求（可以阻止或修改）	阻止危险的 shell 命令
PostToolUse	是	是	工具执行结果	将所有文件更改记录到审计跟踪
PostToolUseFailure	是	是	工具执行失败	处理或记录工具错误
PostToolBatch	否	是	一整批工具调用解决，每批一次，在下一个模型调用之前	为整个批次注入约定
UserPromptSubmit	是	是	用户提示提交	将额外上下文注入到提示中
Stop	是	是	代理执行停止	在退出前保存会话状态
SubagentStart	是	是	子代理初始化	跟踪并行任务生成
SubagentStop	是	是	子代理完成	聚合来自并行任务的结果
PreCompact	是	是	对话压缩请求	在总结前存档完整记录
PermissionRequest	是	是	权限对话将显示	自定义权限处理
SessionStart	否	是	会话初始化	初始化日志记录和遥测
SessionEnd	否	是	会话终止	清理临时资源
Notification	是	是	代理状态消息	将代理状态更新发送到 Slack 或 PagerDuty
Setup	否	是	会话设置/维护	运行初始化任务
TeammateIdle	否	是	队友变为空闲	重新分配工作或通知
TaskCompleted	否	是	后台任务完成	聚合来自并行任务的结果
ConfigChange	否	是	配置文件更改	动态重新加载设置
WorktreeCreate	否	是	Git worktree 创建	跟踪隔离的工作区
WorktreeRemove	否	是	Git worktree 移除	清理工作区资源

配置 hooks

要配置 hook，请在您的代理选项的 hooks 字段中传递它（Python 中的 ClaudeAgentOptions，TypeScript 中的 options 对象）：

Python：

options = ClaudeAgentOptions(
    hooks={"PreToolUse": [HookMatcher(matcher="Bash", hooks=[my_callback])]}
)

async with ClaudeSDKClient(options=options) as client:
    await client.query("Your prompt")
    async for message in client.receive_response():
        print(message)

TypeScript：

for await (const message of query({
  prompt: "Your prompt",
  options: {
    hooks: {
      PreToolUse: [{ matcher: "Bash", hooks: [myCallback] }]
    }
  }
})) {
  console.log(message);
}

hooks 选项是一个字典（Python）或对象（TypeScript），其中：

• 键是 hook 事件名称（例如 'PreToolUse'、'PostToolUse'、'Stop'）
• 值是匹配器数组，每个包含可选的过滤模式和您的回调函数

匹配器

使用匹配器来过滤您的回调何时触发。matcher 字段是一个正则表达式字符串，根据 hook 事件类型匹配不同的值。例如，基于工具的 hooks 匹配工具名称，而 Notification hooks 匹配通知类型。

选项	类型	默认值	描述
matcher	string	undefined	针对事件的过滤字段匹配的正则表达式模式。对于工具 hooks，这是工具名称。内置工具包括 Bash、Read、Write、Edit、Glob、Grep、WebFetch、Agent 等。MCP 工具使用模式 mcp__<server>__<action>。
hooks	HookCallback[]	-	必需。当模式匹配时执行的回调函数数组
timeout	number	60	超时时间（秒）

回调函数

输入

每个 hook 回调接收三个参数：

• 输入数据： 一个包含事件详细信息的类型化对象。每个 hook 类型都有自己的输入形状（例如，PreToolUseHookInput 包括 tool_name 和 tool_input）。
• 工具使用 ID（str | None / string | undefined）：关联同一工具调用的 PreToolUse 和 PostToolUse 事件。
• 上下文： 在 TypeScript 中，包含用于取消的 signal 属性（AbortSignal）。在 Python 中，此参数保留供将来使用。

输出

您的回调返回一个具有两类字段的对象：

• 顶级字段控制对话：systemMessage 将消息注入到对话中，对模型可见，continue（Python 中的 continue_）确定代理在此 hook 后是否继续运行。
• hookSpecificOutput 控制当前操作。内部的字段取决于 hook 事件类型。对于 PreToolUse hooks，这是您设置 permissionDecision（"allow"、"deny" 或 "ask"）、permissionDecisionReason 和 updatedInput 的地方。

返回 {} 以允许操作而不进行更改。

ℹ️ 当多个 hooks 或权限规则适用时，deny 优先于 defer，defer 优先于 ask，ask 优先于 allow。如果任何 hook 返回 deny，操作将被阻止，无论其他 hooks 如何。

异步输出

默认情况下，代理在您的 hook 返回前等待。如果您的 hook 执行副作用（日志记录、发送 webhook）并且不需要影响代理的行为，您可以改为返回异步输出。这告诉代理立即继续，而不等待 hook 完成：

Python：

async def async_hook(input_data, tool_use_id, context):
    asyncio.create_task(send_to_logging_service(input_data))
    return {"async_": True, "asyncTimeout": 30000}

TypeScript：

const asyncHook: HookCallback = async (input, toolUseID, { signal }) => {
  sendToLoggingService(input).catch(console.error);
  return { async: true, asyncTimeout: 30000 };
};

示例

修改工具输入

此示例拦截 Write 工具调用并重写 file_path 参数以添加 /sandbox 前缀，将所有文件写入重定向到沙箱目录。

Python：

async def redirect_to_sandbox(input_data, tool_use_id, context):
    if input_data["hook_event_name"] != "PreToolUse":
        return {}

    if input_data["tool_name"] == "Write":
        original_path = input_data["tool_input"].get("file_path", "")
        return {
            "hookSpecificOutput": {
                "hookEventName": input_data["hook_event_name"],
                "permissionDecision": "allow",
                "updatedInput": {
                    **input_data["tool_input"],
                    "file_path": f"/sandbox{original_path}",
                },
            }
        }
    return {}

TypeScript：

const redirectToSandbox: HookCallback = async (input, toolUseID, { signal }) => {
  if (input.hook_event_name !== "PreToolUse") return {};

  const preInput = input as PreToolUseHookInput;
  const toolInput = preInput.tool_input as Record<string, unknown>;
  if (preInput.tool_name === "Write") {
    const originalPath = toolInput.file_path as string;
    return {
      hookSpecificOutput: {
        hookEventName: preInput.hook_event_name,
        permissionDecision: "allow",
        updatedInput: {
          ...toolInput,
          file_path: `/sandbox${originalPath}`
        }
      }
    };
  }
  return {};
};

ℹ️ 使用 updatedInput 时，您还必须包括 permissionDecision: 'allow'。始终返回新对象而不是改变原始 tool_input。

添加上下文并阻止工具

此示例阻止任何尝试写入 /etc 目录的操作，并将两个输出字段一起使用：permissionDecision: 'deny' 停止工具调用，而 systemMessage 将提醒注入到对话中。

Python：

async def block_etc_writes(input_data, tool_use_id, context):
    file_path = input_data["tool_input"].get("file_path", "")

    if file_path.startswith("/etc"):
        return {
            "systemMessage": "Remember: system directories like /etc are protected.",
            "hookSpecificOutput": {
                "hookEventName": input_data["hook_event_name"],
                "permissionDecision": "deny",
                "permissionDecisionReason": "Writing to /etc is not allowed",
            },
        }
    return {}

TypeScript：

const blockEtcWrites: HookCallback = async (input, toolUseID, { signal }) => {
  const preInput = input as PreToolUseHookInput;
  const toolInput = preInput.tool_input as Record<string, unknown>;
  const filePath = toolInput?.file_path as string;

  if (filePath?.startsWith("/etc")) {
    return {
      systemMessage: "Remember: system directories like /etc are protected.",
      hookSpecificOutput: {
        hookEventName: preInput.hook_event_name,
        permissionDecision: "deny",
        permissionDecisionReason: "Writing to /etc is not allowed"
      }
    };
  }
  return {};
};

自动批准特定工具

默认情况下，代理可能在使用某些工具前提示权限。此示例通过返回 permissionDecision: 'allow' 自动批准只读文件系统工具（Read、Glob、Grep）。

Python：

async def auto_approve_read_only(input_data, tool_use_id, context):
    if input_data["hook_event_name"] != "PreToolUse":
        return {}

    read_only_tools = ["Read", "Glob", "Grep"]
    if input_data["tool_name"] in read_only_tools:
        return {
            "hookSpecificOutput": {
                "hookEventName": input_data["hook_event_name"],
                "permissionDecision": "allow",
                "permissionDecisionReason": "Read-only tool auto-approved",
            }
        }
    return {}

TypeScript：

const autoApproveReadOnly: HookCallback = async (input, toolUseID, { signal }) => {
  if (input.hook_event_name !== "PreToolUse") return {};

  const preInput = input as PreToolUseHookInput;
  const readOnlyTools = ["Read", "Glob", "Grep"];
  if (readOnlyTools.includes(preInput.tool_name)) {
    return {
      hookSpecificOutput: {
        hookEventName: preInput.hook_event_name,
        permissionDecision: "allow",
        permissionDecisionReason: "Read-only tool auto-approved"
      }
    };
  }
  return {};
};

注册多个 hooks

当事件触发时，所有匹配的 hooks 并行运行。对于权限决策，最严格的结果获胜：单个 deny 会阻止工具调用，无论其他 hooks 返回什么。

Python：

options = ClaudeAgentOptions(
    hooks={
        "PreToolUse": [
            HookMatcher(hooks=[authorization_check]),
            HookMatcher(hooks=[input_validator]),
            HookMatcher(hooks=[audit_logger]),
        ]
    }
)

TypeScript：

const options = {
  hooks: {
    PreToolUse: [
      { hooks: [authorizationCheck] },
      { hooks: [inputValidator] },
      { hooks: [auditLogger] }
    ]
  }
};

使用正则表达式匹配器过滤

使用正则表达式模式匹配多个工具：

Python：

options = ClaudeAgentOptions(
    hooks={
        "PreToolUse": [
            # 匹配文件修改工具
            HookMatcher(matcher="Write|Edit|Delete", hooks=[file_security_hook]),
            # 匹配所有 MCP 工具
            HookMatcher(matcher="^mcp__", hooks=[mcp_audit_hook]),
            # 匹配所有内容（无匹配器）
            HookMatcher(hooks=[global_logger]),
        ]
    }
)

TypeScript：

const options = {
  hooks: {
    PreToolUse: [
      // 匹配文件修改工具
      { matcher: "Write|Edit|Delete", hooks: [fileSecurityHook] },

      // 匹配所有 MCP 工具
      { matcher: "^mcp__", hooks: [mcpAuditHook] },

      // 匹配所有内容（无匹配器）
      { hooks: [globalLogger] }
    ]
  }
};

跟踪子代理活动

使用 SubagentStop hooks 监控子代理何时完成其工作：

Python：

async def subagent_tracker(input_data, tool_use_id, context):
    print(f"[SUBAGENT] Completed: {input_data['agent_id']}")
    print(f"  Transcript: {input_data['agent_transcript_path']}")
    print(f"  Tool use ID: {tool_use_id}")
    print(f"  Stop hook active: {input_data.get('stop_hook_active')}")
    return {}


options = ClaudeAgentOptions(
    hooks={"SubagentStop": [HookMatcher(hooks=[subagent_tracker])]}
)

TypeScript：

import { HookCallback, SubagentStopHookInput } from "@anthropic-ai/claude-agent-sdk";

const subagentTracker: HookCallback = async (input, toolUseID, { signal }) => {
  const subInput = input as SubagentStopHookInput;

  console.log(`[SUBAGENT] Completed: ${subInput.agent_id}`);
  console.log(`  Transcript: ${subInput.agent_transcript_path}`);
  console.log(`  Tool use ID: ${toolUseID}`);
  console.log(`  Stop hook active: ${subInput.stop_hook_active}`);
  return {};
};

const options = {
  hooks: {
    SubagentStop: [{ hooks: [subagentTracker] }]
  }
};

从 hooks 发出 HTTP 请求

Hooks 可以执行异步操作，如 HTTP 请求。在您的 hook 内捕获错误，而不是让它们传播。

Python：

import asyncio
import json
import urllib.request
from datetime import datetime


def _send_webhook(tool_name):
    data = json.dumps(
        {
            "tool": tool_name,
            "timestamp": datetime.now().isoformat(),
        }
    ).encode()
    req = urllib.request.Request(
        "https://api.example.com/webhook",
        data=data,
        headers={"Content-Type": "application/json"},
        method="POST",
    )
    urllib.request.urlopen(req)


async def webhook_notifier(input_data, tool_use_id, context):
    if input_data["hook_event_name"] != "PostToolUse":
        return {}

    try:
        await asyncio.to_thread(_send_webhook, input_data["tool_name"])
    except Exception as e:
        print(f"Webhook request failed: {e}")

    return {}

TypeScript：

import { query, HookCallback, PostToolUseHookInput } from "@anthropic-ai/claude-agent-sdk";

const webhookNotifier: HookCallback = async (input, toolUseID, { signal }) => {
  if (input.hook_event_name !== "PostToolUse") return {};

  try {
    await fetch("https://api.example.com/webhook", {
      method: "POST",
      headers: { "Content-Type": "application/json" },
      body: JSON.stringify({
        tool: (input as PostToolUseHookInput).tool_name,
        timestamp: new Date().toISOString()
      }),
      signal
    });
  } catch (error) {
    if (error instanceof Error && error.name === "AbortError") {
      console.log("Webhook request cancelled");
    }
  }

  return {};
};

for await (const message of query({
  prompt: "Refactor the auth module",
  options: {
    hooks: {
      PostToolUse: [{ hooks: [webhookNotifier] }]
    }
  }
})) {
  console.log(message);
}

将通知转发到 Slack

使用 Notification hooks 从代理接收系统通知并将其转发到外部服务。

Python：

import asyncio
import json
import urllib.request

from claude_agent_sdk import ClaudeSDKClient, ClaudeAgentOptions, HookMatcher


def _send_slack_notification(message):
    data = json.dumps({"text": f"Agent status: {message}"}).encode()
    req = urllib.request.Request(
        "https://hooks.slack.com/services/YOUR/WEBHOOK/URL",
        data=data,
        headers={"Content-Type": "application/json"},
        method="POST",
    )
    urllib.request.urlopen(req)


async def notification_handler(input_data, tool_use_id, context):
    try:
        await asyncio.to_thread(_send_slack_notification, input_data.get("message", ""))
    except Exception as e:
        print(f"Failed to send notification: {e}")

    return {}


async def main():
    options = ClaudeAgentOptions(
        hooks={
            "Notification": [HookMatcher(hooks=[notification_handler])],
        },
    )

    async with ClaudeSDKClient(options=options) as client:
        await client.query("Analyze this codebase")
        async for message in client.receive_response():
            print(message)


asyncio.run(main())

TypeScript：

import { query, HookCallback, NotificationHookInput } from "@anthropic-ai/claude-agent-sdk";

const notificationHandler: HookCallback = async (input, toolUseID, { signal }) => {
  const notification = input as NotificationHookInput;

  try {
    await fetch("https://hooks.slack.com/services/YOUR/WEBHOOK/URL", {
      method: "POST",
      headers: { "Content-Type": "application/json" },
      body: JSON.stringify({
        text: `Agent status: ${notification.message}`
      }),
      signal
    });
  } catch (error) {
    if (error instanceof Error && error.name === "AbortError") {
      console.log("Notification cancelled");
    } else {
      console.error("Failed to send notification:", error);
    }
  }

  return {};
};

for await (const message of query({
  prompt: "Analyze this codebase",
  options: {
    hooks: {
      Notification: [{ hooks: [notificationHandler] }]
    }
  }
})) {
  console.log(message);
}

修复常见问题

Hook 未触发

• 验证 hook 事件名称正确且区分大小写（PreToolUse，而不是 preToolUse）
• 检查您的匹配器模式是否与工具名称完全匹配
• 确保 hook 在 options.hooks 中的正确事件类型下
• 对于非工具 hooks，如 Stop 和 SubagentStop，匹配器匹配不同的字段
• 当代理达到 max_turns 限制时，hooks 可能不会触发，因为会话在 hooks 可以执行前结束

匹配器未按预期过滤

匹配器仅匹配工具名称，而不是文件路径或其他参数。要按文件路径过滤，请在您的 hook 内检查 tool_input.file_path：

const myHook: HookCallback = async (input, toolUseID, { signal }) => {
  const preInput = input as PreToolUseHookInput;
  const toolInput = preInput.tool_input as Record<string, unknown>;
  const filePath = toolInput?.file_path as string;
  if (!filePath?.endsWith(".md")) return {}; // 跳过非 markdown 文件
  // 处理 markdown 文件...
  return {};
};

Hook 超时

• 增加 HookMatcher 配置中的 timeout 值
• 在 TypeScript 中使用第三个回调参数中的 AbortSignal 来优雅地处理取消

工具意外被阻止

• 检查所有 PreToolUse hooks 是否返回 permissionDecision: 'deny'
• 向您的 hooks 添加日志记录以查看它们返回的 permissionDecisionReason
• 验证匹配器模式不会太宽泛（空匹配器匹配所有工具）

修改的输入未应用

• 确保 updatedInput 在 hookSpecificOutput 内，而不是在顶级
• 您还必须返回 permissionDecision: 'allow' 以使输入修改生效
• 在 hookSpecificOutput 中包括 hookEventName 以识别输出针对的 hook 类型

Python 中不可用会话 hooks

SessionStart 和 SessionEnd 可以在 TypeScript 中注册为 SDK 回调 hooks，但在 Python SDK 中不可用（HookEvent 省略了它们）。在 Python 中，它们仅作为 shell 命令 hooks 在设置文件中定义。要从您的 SDK 应用程序加载 shell 命令 hooks，请使用 setting_sources 或 settingSources 包括适当的设置源。

子代理权限提示倍增

生成多个子代理时，每个子代理可能会单独请求权限。子代理不会自动继承父代理权限。要避免重复提示，请使用 PreToolUse hooks 自动批准特定工具，或配置适用于子代理会话的权限规则。

子代理的递归 hook 循环

生成子代理的 UserPromptSubmit hook 如果这些子代理触发相同的 hook，可能会创建无限循环。要防止这种情况：

• 在生成子代理前检查 hook 输入中的子代理指示符
• 使用共享变量或会话状态来跟踪您是否已在子代理内
• 将 hooks 范围限制为仅对顶级代理会话运行

systemMessage 未出现在输出中

systemMessage 字段将上下文添加到模型看到的对话中，但它可能不会出现在所有 SDK 输出模式中。如果您需要将 hook 决定呈现给您的应用程序，请单独记录它们或使用专用输出通道。

相关资源

• Claude Code hooks 参考：完整的 JSON 输入/输出架构、事件文档和匹配器模式
• Claude Code hooks 指南：shell 命令 hook 示例和演练
• TypeScript SDK 参考：hook 类型、输入/输出定义和配置选项
• Python SDK 参考：hook 类型、输入/输出定义和配置选项
• 权限：控制您的代理可以做什么
• 自定义工具：构建工具以扩展代理功能

---

本文翻译自 Anthropic Claude Code 官方文档，最近一次同步：2025-05-01。