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

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

  # 定义一个接收工具调用详细信息的 hook 回调
  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]

      # 如果针对 .env 文件，阻止操作
      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 事件注册 hook
              # 匹配器仅过滤 Write 和 Edit 工具调用
              "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())

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

// 使用 HookCallback 类型定义 hook 回调
const protectEnvFiles: HookCallback = async (input, toolUseID, { signal }) => {
  // 将输入转换为特定 hook 类型以获得类型安全
  const preInput = input as PreToolUseHookInput;

  // 转换 tool_input 以访问其属性（在 SDK 中类型为 unknown）
  const toolInput = preInput.tool_input as Record<string, unknown>;
  const filePath = toolInput?.file_path as string;
  const fileName = filePath?.split("/").pop();

  // 如果针对 .env 文件，阻止操作
  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 事件注册 hook
      // 匹配器仅过滤 Write 和 Edit 工具调用
      PreToolUse: [{ matcher: "Write|Edit", hooks: [protectEnvFiles] }]
    }
  }
})) {
  // 过滤助手和结果消息
  if (message.type === "assistant" || message.type === "result") {
    console.log(message);
  }
}

  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)

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

  async def async_hook(input_data, tool_use_id, context):
      # 启动后台任务，然后立即返回
      asyncio.create_task(send_to_logging_service(input_data))
      return {"async_": True, "asyncTimeout": 30000}

const asyncHook: HookCallback = async (input, toolUseID, { signal }) => {
  // 启动后台任务，然后立即返回
  sendToLoggingService(input).catch(console.error);
  return { async: true, asyncTimeout: 30000 };
};

  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 {}

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 {};
};

  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：阻止操作
              "hookSpecificOutput": {
                  "hookEventName": input_data["hook_event_name"],
                  "permissionDecision": "deny",
                  "permissionDecisionReason": "Writing to /etc is not allowed",
              },
          }
      return {}

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：阻止操作
      hookSpecificOutput: {
        hookEventName: preInput.hook_event_name,
        permissionDecision: "deny",
        permissionDecisionReason: "Writing to /etc is not allowed"
      }
    };
  }
  return {};
};

  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 {}

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 {};
};

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

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

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

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

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

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

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

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

const subagentTracker: HookCallback = async (input, toolUseID, { signal }) => {
  // 转换为 SubagentStopHookInput 以访问子代理特定字段
  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] }]
  }
};

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

  def _send_webhook(tool_name):
      """同步辅助函数，将工具使用数据 POST 到外部 webhook。"""
      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):
      # 仅在工具完成后触发（PostToolUse），而不是之前
      if input_data["hook_event_name"] != "PostToolUse":
          return {}

      try:
          # 在线程中运行阻塞 HTTP 调用以避免阻塞事件循环
          await asyncio.to_thread(_send_webhook, input_data["tool_name"])
      except Exception as e:
          # 记录错误但不抛出。失败的 webhook 不应停止代理
          print(f"Webhook request failed: {e}")

      return {}

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

const webhookNotifier: HookCallback = async (input, toolUseID, { signal }) => {
  // 仅在工具完成后触发（PostToolUse），而不是之前
  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 以便在 hook 超时时请求取消
      signal
    });
  } catch (error) {
    // 分别处理取消和其他错误
    if (error instanceof Error && error.name === "AbortError") {
      console.log("Webhook request cancelled");
    }
    // 不重新抛出。失败的 webhook 不应停止代理
  }

  return {};
};

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

  import asyncio
  import json
  import urllib.request

  from claude_agent_sdk import ClaudeSDKClient, ClaudeAgentOptions, HookMatcher

  def _send_slack_notification(message):
      """同步辅助函数，通过传入 webhook 向 Slack 发送消息。"""
      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:
          # 在线程中运行阻塞 HTTP 调用以避免阻塞事件循环
          await asyncio.to_thread(_send_slack_notification, input_data.get("message", ""))
      except Exception as e:
          print(f"Failed to send notification: {e}")

      # 返回空对象。通知 hooks 不修改代理行为
      return {}

  async def main():
      options = ClaudeAgentOptions(
          hooks={
              # 为通知事件注册 hook（不需要匹配器）
              "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())

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

// 定义一个将通知发送到 Slack 的 hook 回调
const notificationHandler: HookCallback = async (input, toolUseID, { signal }) => {
  // 转换为 NotificationHookInput 以访问消息字段
  const notification = input as NotificationHookInput;

  try {
    // 将通知消息 POST 到 Slack 传入 webhook
    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 以便在 hook 超时时请求取消
      signal
    });
  } catch (error) {
    if (error instanceof Error && error.name === "AbortError") {
      console.log("Notification cancelled");
    } else {
      console.error("Failed to send notification:", error);
    }
  }

  // 返回空对象。通知 hooks 不修改代理行为
  return {};
};

// 为通知事件注册 hook（不需要匹配器）
for await (const message of query({
  prompt: "Analyze this codebase",
  options: {
    hooks: {
      Notification: [{ hooks: [notificationHandler] }]
    }
  }
})) {
  console.log(message);
}

  options = ClaudeAgentOptions(
      setting_sources=["project"],  # 加载 .claude/settings.json 包括 hooks
  )

const options = {
  settingSources: ["project"] // 加载 .claude/settings.json 包括 hooks
};