Hooks

HookEvent = (
    Literal["PreToolUse"]
    | Literal["PostToolUse"]
    | Literal["UserPromptSubmit"]
    | Literal["Stop"]
    | Literal["SubagentStop"]
    | Literal["PreCompact"]
)
"""
Supported hook event types.

Hook events allow you to inject custom logic at specific points in Claude's
execution loop:

- PreToolUse: Before a tool is executed
- PostToolUse: After a tool completes execution
- UserPromptSubmit: When user submits a prompt
- Stop: When conversation stops (main agent)
- SubagentStop: When a subagent conversation stops
- PreCompact: Before conversation history is compacted

Note: Due to setup limitations, the Python SDK does not support SessionStart,
SessionEnd, and Notification hooks that are available in the TypeScript SDK.
"""

HookCallback = Callable[
    [dict[str, Any], str | None, HookContext],
    Awaitable[HookJSONOutput]
]
"""
Hook callback function type.

A hook callback is an async function invoked when a hook event occurs.

Args:
    input: Hook input data. Structure varies by hook type:
        - PreToolUse: {"name": tool_name, "input": tool_input}
        - PostToolUse: {"name": tool_name, "output": tool_output}
        - UserPromptSubmit: {"prompt": prompt_text}
        - Stop/SubagentStop: {"result": result_data}
        - PreCompact: {"messages": messages_to_compact}
    tool_use_id: Optional tool use ID (relevant for tool-related hooks)
    context: Hook context with additional information

Returns:
    HookJSONOutput dictionary with optional decision, systemMessage,
    and hookSpecificOutput fields

See https://docs.anthropic.com/en/docs/claude-code/hooks#hook-input
for detailed input structures for each hook type.
"""

@dataclass
class HookContext:
    """
    Context information for hook callbacks.

    Provides additional context and control mechanisms for hooks.
    Future versions may add more fields.

    Attributes:
        signal: Abort signal support (future feature, currently None)
    """

    signal: Any | None = None
    """Future: abort signal support.

    Reserved for future use to allow canceling hook execution.
    Currently always None.
    """

class HookJSONOutput(TypedDict):
    """
    Hook output structure.

    Hooks return this dictionary to control behavior and communicate results.

    Fields:
        decision: Optional. Set to "block" to prevent the action
        systemMessage: Optional. Add a system message to chat transcript
        hookSpecificOutput: Optional. Hook-specific data

    Note: Currently, "continue", "stopReason", and "suppressOutput" from the
    TypeScript SDK are not supported in the Python SDK.

    See https://docs.anthropic.com/en/docs/claude-code/hooks#advanced%3A-json-output
    for detailed documentation.
    """

    decision: NotRequired[Literal["block"]]
    """Whether to block the action related to the hook.

    When set to "block", the action associated with the hook will be prevented:
    - PreToolUse: Block tool execution
    - UserPromptSubmit: Block prompt submission

    If not specified or set to any other value, the action proceeds normally.
    """

    systemMessage: NotRequired[str]
    """Optional system message.

    Add a system message that is saved in the chat transcript but not visible
    to Claude. Useful for logging, debugging, or adding context for later review.
    """

    hookSpecificOutput: NotRequired[Any]
    """Hook-specific output data.

    Each hook type may define specific output fields. See individual hook
    documentation for guidance on what can be included here.
    """

@dataclass
class HookMatcher:
    """
    Hook matcher configuration.

    Defines which events should trigger which hooks. Matchers filter events
    based on patterns like tool names or other criteria.

    Attributes:
        matcher: Optional pattern to filter events
        hooks: List of callback functions to invoke
    """

    matcher: str | None = None
    """Matcher pattern string.

    Filters which events trigger the hooks. The format depends on the hook type:

    For PreToolUse/PostToolUse:
    - Tool name: "Bash" (exact match)
    - Multiple tools: "Write|Edit|MultiEdit" (OR pattern)
    - All tools: None or "" (match any tool)

    For other hooks:
    - Usually None (no filtering)

    See https://docs.anthropic.com/en/docs/claude-code/hooks#structure
    for detailed matcher syntax.
    """

    hooks: list[HookCallback] = field(default_factory=list)
    """List of hook callback functions.

    All callbacks in this list will be invoked when the matcher condition
    is met. Callbacks are executed in order.
    """

from claude_agent_sdk import (
    ClaudeAgentOptions, HookMatcher, HookContext, HookJSONOutput, query
)

async def log_tool_use(input_data: dict, tool_use_id: str | None, context: HookContext) -> HookJSONOutput:
    """Log when tools are used."""
    tool_name = input_data.get("name")
    tool_input = input_data.get("input")
    print(f"Tool called: {tool_name}")
    print(f"Input: {tool_input}")
    return {}

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

async for msg in query(prompt="List files in current directory", options=options):
    print(msg)

from claude_agent_sdk import (
    ClaudeAgentOptions, HookMatcher, HookContext, HookJSONOutput, query
)

async def block_dangerous_bash(
    input_data: dict,
    tool_use_id: str | None,
    context: HookContext
) -> HookJSONOutput:
    """Block dangerous bash commands."""
    tool_input = input_data.get("input", {})
    command = tool_input.get("command", "")

    # Check for dangerous commands
    dangerous_keywords = ["rm -rf", "dd if=", "mkfs", "format", "> /dev/"]

    for keyword in dangerous_keywords:
        if keyword in command:
            return {
                "decision": "block",
                "systemMessage": f"Blocked dangerous command: {command}"
            }

    return {}

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

async for msg in query(prompt="Delete all files", options=options):
    print(msg)

from claude_agent_sdk import (
    ClaudeAgentOptions, HookMatcher, HookContext, HookJSONOutput, query
)

async def validate_write_output(
    input_data: dict,
    tool_use_id: str | None,
    context: HookContext
) -> HookJSONOutput:
    """Validate file write operations."""
    tool_name = input_data.get("name")
    tool_output = input_data.get("output", {})

    if tool_name in ["Write", "Edit"]:
        # Check if write was successful
        is_error = tool_output.get("is_error", False)

        if is_error:
            return {
                "systemMessage": f"File operation failed: {tool_output.get('content')}"
            }
        else:
            return {
                "systemMessage": "File operation completed successfully"
            }

    return {}

options = ClaudeAgentOptions(
    allowed_tools=["Write", "Edit", "Read"],
    hooks={
        "PostToolUse": [
            HookMatcher(matcher="Write|Edit", hooks=[validate_write_output])
        ]
    }
)

async for msg in query(prompt="Create a new Python file", options=options):
    print(msg)

from claude_agent_sdk import (
    ClaudeAgentOptions, HookMatcher, HookContext, HookJSONOutput, query
)

async def log_hook(input_data: dict, tool_use_id: str | None, context: HookContext) -> HookJSONOutput:
    """Log tool usage."""
    print(f"[LOG] Tool: {input_data.get('name')}")
    return {}

async def audit_hook(input_data: dict, tool_use_id: str | None, context: HookContext) -> HookJSONOutput:
    """Audit tool usage."""
    tool_name = input_data.get("name")
    tool_input = input_data.get("input")

    # Save to audit log
    with open("audit.log", "a") as f:
        f.write(f"Tool: {tool_name}, Input: {tool_input}\n")

    return {}

async def security_hook(input_data: dict, tool_use_id: str | None, context: HookContext) -> HookJSONOutput:
    """Security checks."""
    tool_name = input_data.get("name")

    # Block certain tools during business hours
    from datetime import datetime
    hour = datetime.now().hour

    if 9 <= hour <= 17 and tool_name == "Bash":
        return {
            "decision": "block",
            "systemMessage": "Bash commands blocked during business hours"
        }

    return {}

options = ClaudeAgentOptions(
    hooks={
        "PreToolUse": [
            HookMatcher(
                matcher=None,
                hooks=[log_hook, audit_hook, security_hook]
            )
        ]
    }
)

async for msg in query(prompt="Analyze this project", options=options):
    print(msg)

from claude_agent_sdk import (
    ClaudeAgentOptions, HookMatcher, HookContext, HookJSONOutput, query
)

async def validate_prompt(
    input_data: dict,
    tool_use_id: str | None,
    context: HookContext
) -> HookJSONOutput:
    """Validate user prompts before submission."""
    prompt = input_data.get("prompt", "")

    # Block prompts with sensitive information
    sensitive_patterns = ["password", "api_key", "secret", "token"]

    for pattern in sensitive_patterns:
        if pattern.lower() in prompt.lower():
            return {
                "decision": "block",
                "systemMessage": f"Blocked prompt containing sensitive data: {pattern}"
            }

    return {}

options = ClaudeAgentOptions(
    hooks={
        "UserPromptSubmit": [
            HookMatcher(matcher=None, hooks=[validate_prompt])
        ]
    }
)

async for msg in query(prompt="What is my password?", options=options):
    print(msg)

from claude_agent_sdk import (
    ClaudeAgentOptions, HookMatcher, HookContext, HookJSONOutput, query
)

async def cleanup_on_stop(
    input_data: dict,
    tool_use_id: str | None,
    context: HookContext
) -> HookJSONOutput:
    """Clean up resources when conversation stops."""
    result = input_data.get("result", {})

    print(f"Conversation ended")
    print(f"Result: {result}")

    # Perform cleanup
    # - Close database connections
    # - Save state
    # - Log metrics

    return {
        "systemMessage": "Cleanup completed"
    }

options = ClaudeAgentOptions(
    hooks={
        "Stop": [
            HookMatcher(matcher=None, hooks=[cleanup_on_stop])
        ]
    }
)

async for msg in query(prompt="Hello Claude", options=options):
    print(msg)

from datetime import datetime, timedelta
from claude_agent_sdk import (
    ClaudeAgentOptions, HookMatcher, HookContext, HookJSONOutput, query
)

# Rate limiting state
tool_usage = {}
RATE_LIMIT = 5  # Max 5 calls per minute

async def rate_limit_tools(
    input_data: dict,
    tool_use_id: str | None,
    context: HookContext
) -> HookJSONOutput:
    """Rate limit tool usage."""
    tool_name = input_data.get("name")
    now = datetime.now()

    # Initialize tracking for this tool
    if tool_name not in tool_usage:
        tool_usage[tool_name] = []

    # Remove old entries (older than 1 minute)
    tool_usage[tool_name] = [
        ts for ts in tool_usage[tool_name]
        if now - ts < timedelta(minutes=1)
    ]

    # Check rate limit
    if len(tool_usage[tool_name]) >= RATE_LIMIT:
        return {
            "decision": "block",
            "systemMessage": f"Rate limit exceeded for {tool_name} (max {RATE_LIMIT}/min)"
        }

    # Record usage
    tool_usage[tool_name].append(now)

    return {}

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

async for msg in query(prompt="Run many commands", options=options):
    print(msg)

from claude_agent_sdk import (
    ClaudeAgentOptions, HookMatcher, HookContext, HookJSONOutput, query
)

# Application state
class AppState:
    def __init__(self):
        self.readonly_mode = False

app_state = AppState()

async def enforce_readonly(
    input_data: dict,
    tool_use_id: str | None,
    context: HookContext
) -> HookJSONOutput:
    """Block write operations in readonly mode."""
    if not app_state.readonly_mode:
        return {}

    tool_name = input_data.get("name")
    write_tools = ["Write", "Edit", "MultiEdit", "Bash"]

    if tool_name in write_tools:
        return {
            "decision": "block",
            "systemMessage": f"Blocked {tool_name} in readonly mode"
        }

    return {}

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

# Enable readonly mode
app_state.readonly_mode = True

async for msg in query(prompt="Modify this file", options=options):
    print(msg)

import time
from claude_agent_sdk import (
    ClaudeAgentOptions, HookMatcher, HookContext, HookJSONOutput, query
)

# Metrics storage
metrics = {
    "tool_calls": {},
    "tool_durations": {}
}
start_times = {}

async def track_tool_start(
    input_data: dict,
    tool_use_id: str | None,
    context: HookContext
) -> HookJSONOutput:
    """Track tool execution start."""
    tool_name = input_data.get("name")

    # Count tool calls
    metrics["tool_calls"][tool_name] = metrics["tool_calls"].get(tool_name, 0) + 1

    # Record start time
    if tool_use_id:
        start_times[tool_use_id] = time.time()

    return {}

async def track_tool_end(
    input_data: dict,
    tool_use_id: str | None,
    context: HookContext
) -> HookJSONOutput:
    """Track tool execution completion."""
    tool_name = input_data.get("name")

    # Calculate duration
    if tool_use_id and tool_use_id in start_times:
        duration = time.time() - start_times[tool_use_id]

        if tool_name not in metrics["tool_durations"]:
            metrics["tool_durations"][tool_name] = []

        metrics["tool_durations"][tool_name].append(duration)

        del start_times[tool_use_id]

    return {}

async def print_metrics(
    input_data: dict,
    tool_use_id: str | None,
    context: HookContext
) -> HookJSONOutput:
    """Print metrics when conversation stops."""
    print("\n=== Tool Usage Metrics ===")
    print("\nTool Calls:")
    for tool, count in metrics["tool_calls"].items():
        print(f"  {tool}: {count}")

    print("\nAverage Durations:")
    for tool, durations in metrics["tool_durations"].items():
        avg = sum(durations) / len(durations)
        print(f"  {tool}: {avg:.3f}s")

    return {}

options = ClaudeAgentOptions(
    hooks={
        "PreToolUse": [
            HookMatcher(matcher=None, hooks=[track_tool_start])
        ],
        "PostToolUse": [
            HookMatcher(matcher=None, hooks=[track_tool_end])
        ],
        "Stop": [
            HookMatcher(matcher=None, hooks=[print_metrics])
        ]
    }
)

async for msg in query(prompt="Analyze this project", options=options):
    print(msg)

from claude_agent_sdk import (
    ClaudeAgentOptions, HookMatcher, HookContext, HookJSONOutput, query
)

async def custom_tool_output(
    input_data: dict,
    tool_use_id: str | None,
    context: HookContext
) -> HookJSONOutput:
    """Modify tool behavior with hook-specific output."""
    tool_name = input_data.get("name")

    # Example: Add metadata to tool output
    return {
        "systemMessage": f"Tool {tool_name} executed",
        "hookSpecificOutput": {
            "metadata": {
                "tool": tool_name,
                "timestamp": time.time(),
                "source": "hook"
            }
        }
    }

options = ClaudeAgentOptions(
    hooks={
        "PostToolUse": [
            HookMatcher(matcher=None, hooks=[custom_tool_output])
        ]
    }
)

async for msg in query(prompt="List files", options=options):
    print(msg)

from claude_agent_sdk import (
    ClaudeAgentOptions, HookMatcher, HookContext, HookJSONOutput, query
)

async def log_pre_tool(input_data: dict, tool_use_id: str | None, context: HookContext) -> HookJSONOutput:
    print(f"[PRE] Tool: {input_data.get('name')}")
    return {}

async def log_post_tool(input_data: dict, tool_use_id: str | None, context: HookContext) -> HookJSONOutput:
    print(f"[POST] Tool: {input_data.get('name')}")
    return {}

async def log_prompt(input_data: dict, tool_use_id: str | None, context: HookContext) -> HookJSONOutput:
    print(f"[PROMPT] {input_data.get('prompt')}")
    return {}

async def log_stop(input_data: dict, tool_use_id: str | None, context: HookContext) -> HookJSONOutput:
    print("[STOP] Conversation ended")
    return {}

options = ClaudeAgentOptions(
    hooks={
        "PreToolUse": [HookMatcher(matcher=None, hooks=[log_pre_tool])],
        "PostToolUse": [HookMatcher(matcher=None, hooks=[log_post_tool])],
        "UserPromptSubmit": [HookMatcher(matcher=None, hooks=[log_prompt])],
        "Stop": [HookMatcher(matcher=None, hooks=[log_stop])]
    }
)

async for msg in query(prompt="Hello Claude", options=options):
    print(msg)