diff --git a/examples/hooks.py b/examples/hooks.py new file mode 100644 index 0000000..42d782d --- /dev/null +++ b/examples/hooks.py @@ -0,0 +1,184 @@ +#!/usr/bin/env python +"""Example of using hooks with Claude Code SDK via ClaudeCodeOptions. + +This file demonstrates various hook patterns using the hooks parameter +in ClaudeCodeOptions instead of decorator-based hooks. + +Usage: +./examples/hooks.py - List the examples +./examples/hooks.py all - Run all examples +./examples/hooks.py PreToolUse - Run a specific example +""" + +import asyncio +import logging +import sys +from typing import Any + +from claude_code_sdk import ClaudeCodeOptions, ClaudeSDKClient +from claude_code_sdk.types import ( + AssistantMessage, + HookContext, + HookJSONOutput, + HookMatcher, + Message, + ResultMessage, + TextBlock, +) + +# Set up logging to see what's happening +logging.basicConfig(level=logging.INFO, format="%(asctime)s - %(message)s") +logger = logging.getLogger(__name__) + + +def display_message(msg: Message) -> None: + """Standardized message display function.""" + if isinstance(msg, AssistantMessage): + for block in msg.content: + if isinstance(block, TextBlock): + print(f"Claude: {block.text}") + elif isinstance(msg, ResultMessage): + print("Result ended") + + +##### Hook callback functions +async def check_bash_command( + input_data: dict[str, Any], tool_use_id: str | None, context: HookContext +) -> HookJSONOutput: + """Prevent certain bash commands from being executed.""" + tool_name = input_data["tool_name"] + tool_input = input_data["tool_input"] + + if tool_name != "Bash": + return {} + + command = tool_input.get("command", "") + block_patterns = ["foo.sh"] + + for pattern in block_patterns: + if pattern in command: + logger.warning(f"Blocked command: {command}") + return { + "hookSpecificOutput": { + "hookEventName": "PreToolUse", + "permissionDecision": "deny", + "permissionDecisionReason": f"Command contains invalid pattern: {pattern}", + } + } + + return {} + + +async def add_custom_instructions( + input_data: dict[str, Any], tool_use_id: str | None, context: HookContext +) -> HookJSONOutput: + """Add custom instructions when a session starts.""" + return { + "hookSpecificOutput": { + "hookEventName": "SessionStart", + "additionalContext": "My favorite color is hot pink", + } + } + + +async def example_pretooluse() -> None: + """Basic example demonstrating hook protection.""" + print("=== PreToolUse Example ===") + print("This example demonstrates how PreToolUse can block some bash commands but not others.\n") + + # Configure hooks using ClaudeCodeOptions + options = ClaudeCodeOptions( + allowed_tools=["Bash"], + hooks={ + "PreToolUse": [ + HookMatcher(matcher="Bash", hooks=[check_bash_command]), + ], + } + ) + + async with ClaudeSDKClient(options=options) as client: + # Test 1: Command with forbidden pattern (will be blocked) + print("Test 1: Trying a command that our PreToolUse hook should block...") + print("User: Run the bash command: ./foo.sh --help") + await client.query("Run the bash command: ./foo.sh --help") + + async for msg in client.receive_response(): + display_message(msg) + + print("\n" + "=" * 50 + "\n") + + # Test 2: Safe command that should work + print("Test 2: Trying a command that our PreToolUse hook should allow...") + print("User: Run the bash command: echo 'Hello from hooks example!'") + await client.query("Run the bash command: echo 'Hello from hooks example!'") + + async for msg in client.receive_response(): + display_message(msg) + + print("\n" + "=" * 50 + "\n") + + print("\n") + + +async def example_userpromptsubmit() -> None: + """Demonstrate context retention across conversation.""" + print("=== UserPromptSubmit Example ===") + print("This example shows how a UserPromptSubmit hook can add context.\n") + + options = ClaudeCodeOptions( + hooks={ + "UserPromptSubmit": [ + HookMatcher(matcher=None, hooks=[add_custom_instructions]), + ], + } + ) + + async with ClaudeSDKClient(options=options) as client: + print("User: What's my favorite color?") + await client.query("What's my favorite color?") + + async for msg in client.receive_response(): + display_message(msg) + + print("\n") + + +async def main() -> None: + """Run all examples or a specific example based on command line argument.""" + examples = { + "PreToolUse": example_pretooluse, + "UserPromptSubmit": example_userpromptsubmit, + } + + if len(sys.argv) < 2: + # List available examples + print("Usage: python hooks.py ") + print("\nAvailable examples:") + print(" all - Run all examples") + for name in examples: + print(f" {name}") + sys.exit(0) + + example_name = sys.argv[1] + + if example_name == "all": + # Run all examples + for example in examples.values(): + await example() + print("-" * 50 + "\n") + elif example_name in examples: + # Run specific example + await examples[example_name]() + else: + print(f"Error: Unknown example '{example_name}'") + print("\nAvailable examples:") + print(" all - Run all examples") + for name in examples: + print(f" {name}") + sys.exit(1) + + +if __name__ == "__main__": + print("Starting Claude SDK Hooks Examples...") + print("=" * 50 + "\n") + asyncio.run(main()) diff --git a/src/claude_code_sdk/_internal/client.py b/src/claude_code_sdk/_internal/client.py index 037b0bc..03971eb 100644 --- a/src/claude_code_sdk/_internal/client.py +++ b/src/claude_code_sdk/_internal/client.py @@ -6,6 +6,8 @@ from typing import Any from ..types import ( ClaudeCodeOptions, + HookEvent, + HookMatcher, Message, ) from .message_parser import parse_message @@ -21,7 +23,7 @@ class InternalClient: """Initialize the internal client.""" def _convert_hooks_to_internal_format( - self, hooks: dict[str, list[Any]] + self, hooks: dict[HookEvent, list[HookMatcher]] ) -> dict[str, list[dict[str, Any]]]: """Convert HookMatcher format to internal Query format.""" internal_hooks: dict[str, list[dict[str, Any]]] = {} diff --git a/src/claude_code_sdk/client.py b/src/claude_code_sdk/client.py index 48666cd..510d182 100644 --- a/src/claude_code_sdk/client.py +++ b/src/claude_code_sdk/client.py @@ -7,7 +7,7 @@ from dataclasses import replace from typing import Any from ._errors import CLIConnectionError -from .types import ClaudeCodeOptions, Message, ResultMessage +from .types import ClaudeCodeOptions, HookEvent, HookMatcher, Message, ResultMessage class ClaudeSDKClient: @@ -102,7 +102,7 @@ class ClaudeSDKClient: os.environ["CLAUDE_CODE_ENTRYPOINT"] = "sdk-py-client" def _convert_hooks_to_internal_format( - self, hooks: dict[str, list[Any]] + self, hooks: dict[HookEvent, list[HookMatcher]] ) -> dict[str, list[dict[str, Any]]]: """Convert HookMatcher format to internal Query format.""" internal_hooks: dict[str, list[dict[str, Any]]] = {} diff --git a/src/claude_code_sdk/types.py b/src/claude_code_sdk/types.py index 357658d..79d21ff 100644 --- a/src/claude_code_sdk/types.py +++ b/src/claude_code_sdk/types.py @@ -87,7 +87,33 @@ CanUseTool = Callable[ ] -# Hook callback types +##### Hook types +# Supported hook event types. Due to setup limitations, the Python SDK does not +# support SessionStart, SessionEnd, and Notification hooks. +HookEvent = ( + Literal["PreToolUse"] + | Literal["PostToolUse"] + | Literal["UserPromptSubmit"] + | Literal["Stop"] + | Literal["SubagentStop"] + | Literal["PreCompact"] +) + + +# See https://docs.anthropic.com/en/docs/claude-code/hooks#advanced%3A-json-output +# for documentation of the output types. Currently, "continue", "stopReason", +# and "suppressOutput" are not supported in the Python SDK. +class HookJSONOutput(TypedDict): + # Whether to block the action related to the hook. + decision: NotRequired[Literal["block"]] + # Optionally add a system message that is not visible to Claude but saved in + # the chat transcript. + systemMessage: NotRequired[str] + # See each hook's individual "Decision Control" section in the documentation + # for guidance. + hookSpecificOutput: NotRequired[Any] + + @dataclass class HookContext: """Context information for hook callbacks.""" @@ -96,8 +122,14 @@ class HookContext: HookCallback = Callable[ - [dict[str, Any], str | None, HookContext], # input, tool_use_id, context - Awaitable[dict[str, Any]], # response data + # HookCallback input parameters: + # - input + # See https://docs.anthropic.com/en/docs/claude-code/hooks#hook-input for + # the type of 'input', the first value. + # - tool_use_id + # - context + [dict[str, Any], str | None, HookContext], + Awaitable[HookJSONOutput], ] @@ -106,8 +138,14 @@ HookCallback = Callable[ class HookMatcher: """Hook matcher configuration.""" - matcher: dict[str, Any] | None = None # Matcher criteria - hooks: list[HookCallback] = field(default_factory=list) # Callbacks to invoke + # See https://docs.anthropic.com/en/docs/claude-code/hooks#structure for the + # expected string value. For example, for PreToolUse, the matcher can be + # a tool name like "Bash" or a combination of tool names like + # "Write|MultiEdit|Edit". + matcher: str | None = None + + # A list of Python functions with function signature HookCallback + hooks: list[HookCallback] = field(default_factory=list) # MCP Server config @@ -259,7 +297,7 @@ class ClaudeCodeOptions: can_use_tool: CanUseTool | None = None # Hook configurations - hooks: dict[str, list[HookMatcher]] | None = None + hooks: dict[HookEvent, list[HookMatcher]] | None = None # SDK Control Protocol @@ -278,8 +316,7 @@ class SDKControlPermissionRequest(TypedDict): class SDKControlInitializeRequest(TypedDict): subtype: Literal["initialize"] - # TODO: Use HookEvent names as the key. - hooks: dict[str, Any] | None + hooks: dict[HookEvent, Any] | None class SDKControlSetPermissionModeRequest(TypedDict):