Claude Agent SDK Hooks: TypeScript and Python Examples

import type { HookCallback, Options } from "@anthropic-ai/claude-agent-sdk";
 
const patterns = [
  /DELETE_DEMO/, /\brm\s+-rf\b/, /(?:^|\s)sudo(?:\s|$)/,
  /\bcurl\b[\s\S]*\|\s*(?:sh|bash)\b/,
];
export const blockCommand: HookCallback = async (input) => {
  if (input.hook_event_name !== "PreToolUse" || input.tool_name !== "Bash") return {};
  const args = input.tool_input as Record<string, unknown>;
  const command = typeof args.command === "string" ? args.command : "";
  if (!patterns.some((pattern) => pattern.test(command))) return {};
  console.log({ tool_use_id: input.tool_use_id, decision: "deny" });
  return { hookSpecificOutput: {
    hookEventName: "PreToolUse",
    permissionDecision: "deny",
    permissionDecisionReason: "Command matched the application's block list.",
  } };
};
export const hooks: NonNullable<Options["hooks"]> = {
  PreToolUse: [{ matcher: "Bash", timeout: 5, hooks: [blockCommand] }],
};

import re
from claude_agent_sdk import HookMatcher
 
PATTERNS = [
    r"DELETE_DEMO", r"\brm\s+-rf\b", r"(?:^|\s)sudo(?:\s|$)",
    r"\bcurl\b[\s\S]*\|\s*(?:sh|bash)\b",
]
 
async def block_command(input_data, tool_use_id, context):
    if input_data["hook_event_name"] != "PreToolUse" or input_data["tool_name"] != "Bash":
        return {}
    command = input_data["tool_input"].get("command", "")
    if not isinstance(command, str) or not any(re.search(p, command) for p in PATTERNS):
        return {}
    print({"tool_use_id": input_data["tool_use_id"], "decision": "deny"})
    return {"hookSpecificOutput": {
        "hookEventName": "PreToolUse", "permissionDecision": "deny",
        "permissionDecisionReason": "Command matched the application's block list.",
    }}
 
hooks = {"PreToolUse": [HookMatcher(matcher="Bash", timeout=5, hooks=[block_command])]}

import { query } from "@anthropic-ai/claude-agent-sdk";
import { hooks } from "./hooks.js";
 
const tools: string[] = ["Bash"];
const prompt = "Use Bash to run exactly: echo DELETE_DEMO. If denied, explain without retrying.";
 
for await (const message of query({
  prompt,
  options: {
    cwd: process.cwd(),
    tools,
    allowedTools: tools,
    permissionMode: "dontAsk",
    settingSources: [],
    strictMcpConfig: true,
    maxTurns: 6,
    hooks,
  },
})) {
  if (message.type !== "result") continue;
  console.log({ subtype: message.subtype, is_error: message.is_error,
    terminal_reason: message.terminal_reason });
  if (message.subtype === "success") console.log(message.result);
}

import asyncio
from pathlib import Path
from claude_agent_sdk import ClaudeAgentOptions, ClaudeSDKClient, ResultMessage
from hooks import hooks
 
TOOLS = ["Bash"]
PROMPT = "Use Bash to run exactly: echo DELETE_DEMO. If denied, explain without retrying."
 
async def main():
    options = ClaudeAgentOptions(
        cwd=str(Path.cwd()), tools=TOOLS, allowed_tools=TOOLS,
        permission_mode="dontAsk", setting_sources=[], strict_mcp_config=True,
        max_turns=6, hooks=hooks,
    )
    async with ClaudeSDKClient(options=options) as client:
        await client.query(PROMPT)
        async for message in client.receive_response():
            if isinstance(message, ResultMessage):
                print({"subtype": message.subtype, "is_error": message.is_error,
                       "terminal_reason": message.terminal_reason})
                if message.subtype == "success":
                    print(message.result)
 
if __name__ == "__main__":
    asyncio.run(main())

// HookCallback signature:
// (input: HookInput, toolUseID: string | undefined,
//  options: { signal: AbortSignal }) => Promise<HookJSONOutput>

from claude_agent_sdk import HookInput, HookContext, HookJSONOutput
 
async def callback(input_data: HookInput, tool_use_id: str | None,
                   context: HookContext) -> HookJSONOutput:
    return {}

import { appendFile } from "node:fs/promises";
import type { HookCallback, Options } from "@anthropic-ai/claude-agent-sdk";
 
let queue: Promise<void> = Promise.resolve();
const audit: HookCallback = async (input) => {
  if (input.hook_event_name !== "PreToolUse" &&
      input.hook_event_name !== "PostToolUse" &&
      input.hook_event_name !== "PostToolUseFailure") return {};
  const line = JSON.stringify({
    event: input.hook_event_name, session_id: input.session_id,
    agent_id: input.agent_id ?? null,
    tool_name: input.tool_name, tool_use_id: input.tool_use_id,
  }) + "\n";
  const write = queue.then(() => appendFile("./audit.jsonl", line));
  queue = write.catch(() => {});
  try { await write; } catch (error) { console.error(error); }
  return {};
};
export const hooks: NonNullable<Options["hooks"]> = {
  PreToolUse: [{ timeout: 5, hooks: [audit] }],
  PostToolUse: [{ timeout: 5, hooks: [audit] }],
  PostToolUseFailure: [{ timeout: 5, hooks: [audit] }],
};

import asyncio
import json
from claude_agent_sdk import HookMatcher
 
lock = asyncio.Lock()
 
def append(line):
    with open("./audit.jsonl", "a", encoding="utf-8") as file:
        file.write(line)
 
async def audit(input_data, tool_use_id, context):
    if input_data["hook_event_name"] not in (
        "PreToolUse", "PostToolUse", "PostToolUseFailure"
    ):
        return {}
    record = {key: input_data.get(key) for key in (
        "hook_event_name", "session_id", "agent_id", "tool_name", "tool_use_id"
    )}
    try:
        async with lock:
            await asyncio.to_thread(append, json.dumps(record) + "\n")
    except OSError as error:
        print(error)
    return {}
 
hooks = {event: [HookMatcher(timeout=5, hooks=[audit])] for event in (
    "PreToolUse", "PostToolUse", "PostToolUseFailure"
)}

import type { HookCallback, Options } from "@anthropic-ai/claude-agent-sdk";
 
const redact: HookCallback = async (input) => {
  if (input.hook_event_name !== "PreToolUse") return {};
  const args = input.tool_input as Record<string, unknown>;
  const field = input.tool_name === "Write" ? "content" :
    input.tool_name === "Edit" ? "new_string" : null;
  if (!field || typeof args[field] !== "string") return {};
  const after = (args[field] as string).split("DEMO_SECRET_123").join("[REDACTED]");
  if (after === args[field]) return {};
  return { hookSpecificOutput: {
    hookEventName: "PreToolUse", updatedInput: { ...args, [field]: after },
    additionalContext: "A known secret was removed from this write.",
  } };
};
export const hooks: NonNullable<Options["hooks"]> = {
  PreToolUse: [{ matcher: "Write|Edit", timeout: 5, hooks: [redact] }],
};

from claude_agent_sdk import HookMatcher
 
async def redact(input_data, tool_use_id, context):
    if input_data["hook_event_name"] != "PreToolUse":
        return {}
    field = {"Write": "content", "Edit": "new_string"}.get(input_data["tool_name"])
    args = input_data["tool_input"]
    if field is None or not isinstance(args.get(field), str):
        return {}
    after = args[field].replace("DEMO_SECRET_123", "[REDACTED]")
    if after == args[field]:
        return {}
    return {"hookSpecificOutput": {
        "hookEventName": "PreToolUse", "updatedInput": {**args, field: after},
        "additionalContext": "A known secret was removed from this write.",
    }}
 
hooks = {"PreToolUse": [HookMatcher(matcher="Write|Edit", timeout=5, hooks=[redact])]}

import path from "node:path";
import { lstat, realpath } from "node:fs/promises";
import type { HookCallback, Options } from "@anthropic-ai/claude-agent-sdk";
 
const root = await realpath("./workspace");
function inside(target: string) {
  const rel = path.relative(root, target);
  return !path.isAbsolute(rel) && rel !== ".." && !rel.startsWith(`..${path.sep}`);
}
async function permitted(file: string) {
  try {
    if (!path.isAbsolute(file) || file.split(/[\\/]/).includes("..") || /[\\/]$/.test(file)) return false;
    if (!inside(await realpath(path.dirname(file)))) return false;
    let entry;
    try { entry = await lstat(file); }
    catch (error) { return (error as NodeJS.ErrnoException).code === "ENOENT"; }
    return entry.isFile() && !entry.isSymbolicLink() && inside(await realpath(file));
  } catch { return false; }
}
const guard: HookCallback = async (input) => {
  if (input.hook_event_name !== "PreToolUse") return {};
  const args = input.tool_input as Record<string, unknown>;
  if (typeof args.file_path === "string" && await permitted(args.file_path)) return {};
  return { hookSpecificOutput: {
    hookEventName: "PreToolUse", permissionDecision: "deny",
    permissionDecisionReason: "Write must stay inside the allowed workspace.",
  } };
};
export const hooks: NonNullable<Options["hooks"]> = {
  PreToolUse: [{ matcher: "Write|Edit", timeout: 5, hooks: [guard] }],
};

import asyncio
import re
import stat
from pathlib import Path
from claude_agent_sdk import HookMatcher
 
ROOT = Path("./workspace").resolve(strict=True)
 
def permitted(file):
    try:
        target = Path(file)
        if not target.is_absolute() or ".." in re.split(r"[\\/]", file) or re.search(r"[\\/]$", file):
            return False
        if not target.parent.resolve(strict=True).is_relative_to(ROOT):
            return False
        try:
            mode = target.lstat().st_mode
        except FileNotFoundError:
            return True
        return stat.S_ISREG(mode) and target.resolve(strict=True).is_relative_to(ROOT)
    except (OSError, ValueError, RuntimeError):
        return False
 
async def guard(input_data, tool_use_id, context):
    if input_data["hook_event_name"] != "PreToolUse":
        return {}
    file = input_data["tool_input"].get("file_path")
    if isinstance(file, str) and await asyncio.to_thread(permitted, file):
        return {}
    return {"hookSpecificOutput": {
        "hookEventName": "PreToolUse", "permissionDecision": "deny",
        "permissionDecisionReason": "Write must stay inside the allowed workspace.",
    }}
 
hooks = {"PreToolUse": [HookMatcher(matcher="Write|Edit", timeout=5, hooks=[guard])]}

import type { HookCallback, Options } from "@anthropic-ai/claude-agent-sdk";
 
const context: HookCallback = async (input) => {
  if (input.hook_event_name !== "UserPromptSubmit") return {};
  return { hookSpecificOutput: {
    hookEventName: "UserPromptSubmit",
    additionalContext: "Environment: staging. Test command: python3 -m unittest discover -s tests.",
  } };
};
export const hooks: NonNullable<Options["hooks"]> = {
  UserPromptSubmit: [{ timeout: 5, hooks: [context] }],
};

from claude_agent_sdk import HookMatcher
 
async def add_context(input_data, tool_use_id, context):
    if input_data["hook_event_name"] != "UserPromptSubmit":
        return {}
    return {"hookSpecificOutput": {
        "hookEventName": "UserPromptSubmit",
        "additionalContext": "Environment: staging. Test command: python3 -m unittest discover -s tests.",
    }}
 
hooks = {"UserPromptSubmit": [HookMatcher(timeout=5, hooks=[add_context])]}

import { readFile } from "node:fs/promises";
import path from "node:path";
import type { HookCallback, Options } from "@anthropic-ai/claude-agent-sdk";
 
const summary = path.resolve("workspace/SUMMARY.md");
export async function summaryReady() {
  try { return (await readFile(summary, "utf8")).trim().length > 0; }
  catch { return false; }
}
const requireSummary: HookCallback = async (input) => {
  if (input.hook_event_name !== "Stop" || input.stop_hook_active) return {};
  if (await summaryReady()) return {};
  return { decision: "block", reason: `Write a nonempty summary to ${summary} before finishing.` };
};
export const hooks: NonNullable<Options["hooks"]> = {
  Stop: [{ timeout: 5, hooks: [requireSummary] }],
};

if (!(await summaryReady())) throw new Error("Summary missing or empty");

import asyncio
from pathlib import Path
from claude_agent_sdk import HookMatcher
 
SUMMARY = Path("workspace/SUMMARY.md").resolve()
 
async def summary_ready():
    try:
        text = await asyncio.to_thread(SUMMARY.read_text, encoding="utf-8")
        return bool(text.strip())
    except (OSError, UnicodeError):
        return False
 
async def require_summary(input_data, tool_use_id, context):
    if input_data["hook_event_name"] != "Stop" or input_data["stop_hook_active"]:
        return {}
    if await summary_ready():
        return {}
    return {"decision": "block", "reason": f"Write a nonempty summary to {SUMMARY} before finishing."}
 
hooks = {"Stop": [HookMatcher(timeout=5, hooks=[require_summary])]}

    if not await summary_ready():
        raise RuntimeError("Summary missing or empty")

import type { SDKResultMessage } from "@anthropic-ai/claude-agent-sdk";
 
export const hooks = {};
export function logUsage(message: SDKResultMessage) {
  console.log({
    session_id: message.session_id,
    estimated_cost_usd: message.total_cost_usd,
    model_usage: message.modelUsage,
  });
}

from claude_agent_sdk import ResultMessage
 
hooks = {}
 
def log_usage(message: ResultMessage):
    print({
        "session_id": message.session_id,
        "estimated_cost_usd": message.total_cost_usd,
        "model_usage": message.model_usage,
    })

import assert from "node:assert/strict";
import { blockCommand } from "./hooks.js";
 
const output = await blockCommand({
  hook_event_name: "PreToolUse", session_id: "test", cwd: process.cwd(),
  transcript_path: "/tmp/fixture.jsonl", tool_name: "Bash",
  tool_input: { command: "echo DELETE_DEMO" }, tool_use_id: "test-tool",
}, "test-tool", { signal: new AbortController().signal });
assert("hookSpecificOutput" in output &&
  output.hookSpecificOutput?.hookEventName === "PreToolUse");
assert.equal(output.hookSpecificOutput.permissionDecision, "deny");

npx tsx hooks.test.ts
npx tsc --noEmit --strict --target ES2022 --module NodeNext \
  --moduleResolution NodeNext --types node hooks.ts agent.ts hooks.test.ts

import asyncio
from hooks import block_command
 
async def test():
    output = await block_command({
        "hook_event_name": "PreToolUse", "tool_name": "Bash",
        "tool_input": {"command": "echo DELETE_DEMO"}, "tool_use_id": "test-tool",
        "session_id": "test", "cwd": ".", "transcript_path": "/tmp/fixture.jsonl",
    }, "test-tool", {"signal": None})
    assert output["hookSpecificOutput"]["permissionDecision"] == "deny"
 
asyncio.run(test())