diff --git a/.github/workflows/test.yml b/.github/workflows/test.yml index 9822a8fe..75eafa27 100644 --- a/.github/workflows/test.yml +++ b/.github/workflows/test.yml @@ -125,6 +125,8 @@ jobs: run: | python examples/quick_start.py timeout 120 python examples/streaming_mode.py all + timeout 120 python examples/hooks.py PreToolUse + timeout 120 python examples/hooks.py DecisionFields - name: Run example scripts (Windows) if: runner.os == 'Windows' @@ -136,4 +138,14 @@ jobs: Wait-Job $job -Timeout 120 | Out-Null Stop-Job $job Receive-Job $job + + $job = Start-Job { python examples/hooks.py PreToolUse } + Wait-Job $job -Timeout 120 | Out-Null + Stop-Job $job + Receive-Job $job + + $job = Start-Job { python examples/hooks.py DecisionFields } + Wait-Job $job -Timeout 120 | Out-Null + Stop-Job $job + Receive-Job $job shell: pwsh diff --git a/e2e-tests/test_hooks.py b/e2e-tests/test_hooks.py new file mode 100644 index 00000000..a01b65c0 --- /dev/null +++ b/e2e-tests/test_hooks.py @@ -0,0 +1,149 @@ +"""End-to-end tests for hook callbacks with real Claude API calls.""" + +import pytest + +from claude_agent_sdk import ( + ClaudeAgentOptions, + ClaudeSDKClient, + HookContext, + HookJSONOutput, + HookMatcher, +) + + +@pytest.mark.e2e +@pytest.mark.asyncio +async def test_hook_with_permission_decision_and_reason(): + """Test that hooks with permissionDecision and reason fields work end-to-end.""" + hook_invocations = [] + + async def test_hook( + input_data: dict, tool_use_id: str | None, context: HookContext + ) -> HookJSONOutput: + """Hook that uses permissionDecision and reason fields.""" + tool_name = input_data.get("tool_name", "") + print(f"Hook called for tool: {tool_name}") + hook_invocations.append(tool_name) + + # Block Bash commands for this test + if tool_name == "Bash": + return { + "reason": "Bash commands are blocked in this test for safety", + "systemMessage": "⚠️ Command blocked by hook", + "hookSpecificOutput": { + "hookEventName": "PreToolUse", + "permissionDecision": "deny", + "permissionDecisionReason": "Security policy: Bash blocked", + }, + } + + return { + "reason": "Tool approved by security review", + "hookSpecificOutput": { + "hookEventName": "PreToolUse", + "permissionDecision": "allow", + "permissionDecisionReason": "Tool passed security checks", + }, + } + + options = ClaudeAgentOptions( + allowed_tools=["Bash", "Write"], + hooks={ + "PreToolUse": [ + HookMatcher(matcher="Bash", hooks=[test_hook]), + ], + }, + ) + + async with ClaudeSDKClient(options=options) as client: + await client.query("Run this bash command: echo 'hello'") + + async for message in client.receive_response(): + print(f"Got message: {message}") + + print(f"Hook invocations: {hook_invocations}") + # Verify hook was called + assert "Bash" in hook_invocations, f"Hook should have been invoked for Bash tool, got: {hook_invocations}" + + +@pytest.mark.e2e +@pytest.mark.asyncio +async def test_hook_with_continue_and_stop_reason(): + """Test that hooks with continue_=False and stopReason fields work end-to-end.""" + hook_invocations = [] + + async def post_tool_hook( + input_data: dict, tool_use_id: str | None, context: HookContext + ) -> HookJSONOutput: + """PostToolUse hook that stops execution with stopReason.""" + tool_name = input_data.get("tool_name", "") + hook_invocations.append(tool_name) + + # Actually test continue_=False and stopReason fields + return { + "continue_": False, + "stopReason": "Execution halted by test hook for validation", + "reason": "Testing continue and stopReason fields", + "systemMessage": "🛑 Test hook stopped execution", + } + + options = ClaudeAgentOptions( + allowed_tools=["Bash"], + hooks={ + "PostToolUse": [ + HookMatcher(matcher="Bash", hooks=[post_tool_hook]), + ], + }, + ) + + async with ClaudeSDKClient(options=options) as client: + await client.query("Run: echo 'test message'") + + async for message in client.receive_response(): + print(f"Got message: {message}") + + print(f"Hook invocations: {hook_invocations}") + # Verify hook was called + assert "Bash" in hook_invocations, f"PostToolUse hook should have been invoked, got: {hook_invocations}" + + +@pytest.mark.e2e +@pytest.mark.asyncio +async def test_hook_with_additional_context(): + """Test that hooks with hookSpecificOutput work end-to-end.""" + hook_invocations = [] + + async def context_hook( + input_data: dict, tool_use_id: str | None, context: HookContext + ) -> HookJSONOutput: + """Hook that provides additional context.""" + hook_invocations.append("context_added") + + return { + "systemMessage": "Additional context provided by hook", + "reason": "Hook providing monitoring feedback", + "suppressOutput": False, + "hookSpecificOutput": { + "hookEventName": "PostToolUse", + "additionalContext": "The command executed successfully with hook monitoring", + }, + } + + options = ClaudeAgentOptions( + allowed_tools=["Bash"], + hooks={ + "PostToolUse": [ + HookMatcher(matcher="Bash", hooks=[context_hook]), + ], + }, + ) + + async with ClaudeSDKClient(options=options) as client: + await client.query("Run: echo 'testing hooks'") + + async for message in client.receive_response(): + print(f"Got message: {message}") + + print(f"Hook invocations: {hook_invocations}") + # Verify hook was called + assert "context_added" in hook_invocations, "Hook with hookSpecificOutput should have been invoked" diff --git a/examples/hooks.py b/examples/hooks.py index 18bd8815..e533ac7d 100644 --- a/examples/hooks.py +++ b/examples/hooks.py @@ -81,6 +81,77 @@ async def add_custom_instructions( } +async def review_tool_output( + input_data: dict[str, Any], tool_use_id: str | None, context: HookContext +) -> HookJSONOutput: + """Review tool output and provide additional context or warnings.""" + tool_response = input_data.get("tool_response", "") + + # If the tool produced an error, add helpful context + if "error" in str(tool_response).lower(): + return { + "systemMessage": "⚠️ The command produced an error", + "reason": "Tool execution failed - consider checking the command syntax", + "hookSpecificOutput": { + "hookEventName": "PostToolUse", + "additionalContext": "The command encountered an error. You may want to try a different approach.", + } + } + + return {} + + +async def strict_approval_hook( + input_data: dict[str, Any], tool_use_id: str | None, context: HookContext +) -> HookJSONOutput: + """Demonstrates using permissionDecision to control tool execution.""" + tool_name = input_data.get("tool_name") + tool_input = input_data.get("tool_input", {}) + + # Block any Write operations to specific files + if tool_name == "Write": + file_path = tool_input.get("file_path", "") + if "important" in file_path.lower(): + logger.warning(f"Blocked Write to: {file_path}") + return { + "reason": "Writes to files containing 'important' in the name are not allowed for safety", + "systemMessage": "🚫 Write operation blocked by security policy", + "hookSpecificOutput": { + "hookEventName": "PreToolUse", + "permissionDecision": "deny", + "permissionDecisionReason": "Security policy blocks writes to important files", + }, + } + + # Allow everything else explicitly + return { + "reason": "Tool use approved after security review", + "hookSpecificOutput": { + "hookEventName": "PreToolUse", + "permissionDecision": "allow", + "permissionDecisionReason": "Tool passed security checks", + }, + } + + +async def stop_on_error_hook( + input_data: dict[str, Any], tool_use_id: str | None, context: HookContext +) -> HookJSONOutput: + """Demonstrates using continue=False to stop execution on certain conditions.""" + tool_response = input_data.get("tool_response", "") + + # Stop execution if we see a critical error + if "critical" in str(tool_response).lower(): + logger.error("Critical error detected - stopping execution") + return { + "continue_": False, + "stopReason": "Critical error detected in tool output - execution halted for safety", + "systemMessage": "🛑 Execution stopped due to critical error", + } + + return {"continue_": True} + + async def example_pretooluse() -> None: """Basic example demonstrating hook protection.""" print("=== PreToolUse Example ===") @@ -143,11 +214,99 @@ async def example_userpromptsubmit() -> None: print("\n") +async def example_posttooluse() -> None: + """Demonstrate PostToolUse hook with reason and systemMessage fields.""" + print("=== PostToolUse Example ===") + print("This example shows how PostToolUse can provide feedback with reason and systemMessage.\n") + + options = ClaudeAgentOptions( + allowed_tools=["Bash"], + hooks={ + "PostToolUse": [ + HookMatcher(matcher="Bash", hooks=[review_tool_output]), + ], + } + ) + + async with ClaudeSDKClient(options=options) as client: + print("User: Run a command that will produce an error: ls /nonexistent_directory") + await client.query("Run this command: ls /nonexistent_directory") + + async for msg in client.receive_response(): + display_message(msg) + + print("\n") + + +async def example_decision_fields() -> None: + """Demonstrate permissionDecision, reason, and systemMessage fields.""" + print("=== Permission Decision Example ===") + print("This example shows how to use permissionDecision='allow'/'deny' with reason and systemMessage.\n") + + options = ClaudeAgentOptions( + allowed_tools=["Write", "Bash"], + model="claude-sonnet-4-5-20250929", + hooks={ + "PreToolUse": [ + HookMatcher(matcher="Write", hooks=[strict_approval_hook]), + ], + } + ) + + async with ClaudeSDKClient(options=options) as client: + # Test 1: Try to write to a file with "important" in the name (should be blocked) + print("Test 1: Trying to write to important_config.txt (should be blocked)...") + print("User: Write 'test' to important_config.txt") + await client.query("Write the text 'test data' to a file called important_config.txt") + + async for msg in client.receive_response(): + display_message(msg) + + print("\n" + "=" * 50 + "\n") + + # Test 2: Write to a regular file (should be approved) + print("Test 2: Trying to write to regular_file.txt (should be approved)...") + print("User: Write 'test' to regular_file.txt") + await client.query("Write the text 'test data' to a file called regular_file.txt") + + async for msg in client.receive_response(): + display_message(msg) + + print("\n") + + +async def example_continue_control() -> None: + """Demonstrate continue and stopReason fields for execution control.""" + print("=== Continue/Stop Control Example ===") + print("This example shows how to use continue_=False with stopReason to halt execution.\n") + + options = ClaudeAgentOptions( + allowed_tools=["Bash"], + hooks={ + "PostToolUse": [ + HookMatcher(matcher="Bash", hooks=[stop_on_error_hook]), + ], + } + ) + + async with ClaudeSDKClient(options=options) as client: + print("User: Run a command that outputs 'CRITICAL ERROR'") + await client.query("Run this bash command: echo 'CRITICAL ERROR: system failure'") + + 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, + "PostToolUse": example_posttooluse, + "DecisionFields": example_decision_fields, + "ContinueControl": example_continue_control, } if len(sys.argv) < 2: @@ -157,6 +316,12 @@ async def main() -> None: print(" all - Run all examples") for name in examples: print(f" {name}") + print("\nExample descriptions:") + print(" PreToolUse - Block commands using PreToolUse hook") + print(" UserPromptSubmit - Add context at prompt submission") + print(" PostToolUse - Review tool output with reason and systemMessage") + print(" DecisionFields - Use permissionDecision='allow'/'deny' with reason") + print(" ContinueControl - Control execution with continue_ and stopReason") sys.exit(0) example_name = sys.argv[1] diff --git a/src/claude_agent_sdk/__init__.py b/src/claude_agent_sdk/__init__.py index 01f5a576..45eccec4 100644 --- a/src/claude_agent_sdk/__init__.py +++ b/src/claude_agent_sdk/__init__.py @@ -23,6 +23,7 @@ ContentBlock, HookCallback, HookContext, + HookJSONOutput, HookMatcher, McpSdkServerConfig, McpServerConfig, @@ -308,6 +309,7 @@ async def call_tool(name: str, arguments: dict[str, Any]) -> Any: "PermissionUpdate", "HookCallback", "HookContext", + "HookJSONOutput", "HookMatcher", # Agent support "AgentDefinition", diff --git a/src/claude_agent_sdk/types.py b/src/claude_agent_sdk/types.py index 4f9c27dd..a1cb210a 100644 --- a/src/claude_agent_sdk/types.py +++ b/src/claude_agent_sdk/types.py @@ -157,18 +157,74 @@ class PermissionResultDeny: ) +# Hook-specific output types +class PreToolUseHookSpecificOutput(TypedDict): + """Hook-specific output for PreToolUse events.""" + + hookEventName: Literal["PreToolUse"] + permissionDecision: NotRequired[Literal["allow", "deny", "ask"]] + permissionDecisionReason: NotRequired[str] + updatedInput: NotRequired[dict[str, Any]] + + +class PostToolUseHookSpecificOutput(TypedDict): + """Hook-specific output for PostToolUse events.""" + + hookEventName: Literal["PostToolUse"] + additionalContext: NotRequired[str] + + +class UserPromptSubmitHookSpecificOutput(TypedDict): + """Hook-specific output for UserPromptSubmit events.""" + + hookEventName: Literal["UserPromptSubmit"] + additionalContext: NotRequired[str] + + +class SessionStartHookSpecificOutput(TypedDict): + """Hook-specific output for SessionStart events.""" + + hookEventName: Literal["SessionStart"] + additionalContext: NotRequired[str] + + +HookSpecificOutput = ( + PreToolUseHookSpecificOutput + | PostToolUseHookSpecificOutput + | UserPromptSubmitHookSpecificOutput + | SessionStartHookSpecificOutput +) + + # 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. +# for documentation of the output types. +class AsyncHookJSONOutput(TypedDict): + """Async hook output that defers hook execution.""" + + async_: Literal[True] # Using async_ to avoid Python keyword + asyncTimeout: NotRequired[int] + + +class SyncHookJSONOutput(TypedDict): + """Synchronous hook output with control and decision fields.""" + + # Common control fields + continue_: NotRequired[bool] # Using continue_ to avoid Python keyword + suppressOutput: NotRequired[bool] + stopReason: NotRequired[str] + + # Decision fields + # Note: "approve" is deprecated for PreToolUse (use permissionDecision instead) + # For other hooks, only "block" is meaningful 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] + reason: NotRequired[str] + + # Hook-specific outputs + hookSpecificOutput: NotRequired[HookSpecificOutput] + + +HookJSONOutput = AsyncHookJSONOutput | SyncHookJSONOutput @dataclass diff --git a/tests/test_tool_callbacks.py b/tests/test_tool_callbacks.py index 8e69fc5e..789f4204 100644 --- a/tests/test_tool_callbacks.py +++ b/tests/test_tool_callbacks.py @@ -1,10 +1,13 @@ """Tests for tool permission callbacks and hook callbacks.""" +import json + import pytest from claude_agent_sdk import ( ClaudeAgentOptions, HookContext, + HookJSONOutput, HookMatcher, PermissionResultAllow, PermissionResultDeny, @@ -257,6 +260,136 @@ async def test_hook( last_response = transport.written_messages[-1] assert '"processed": true' in last_response + @pytest.mark.asyncio + async def test_hook_output_fields(self): + """Test that all SyncHookJSONOutput fields are properly handled.""" + + # Test all SyncHookJSONOutput fields together + async def comprehensive_hook( + input_data: dict, tool_use_id: str | None, context: HookContext + ) -> HookJSONOutput: + return { + # Control fields + "continue_": True, + "suppressOutput": False, + "stopReason": "Test stop reason", + # Decision fields + "decision": "block", + "systemMessage": "Test system message", + "reason": "Test reason for blocking", + # Hook-specific output with all PreToolUse fields + "hookSpecificOutput": { + "hookEventName": "PreToolUse", + "permissionDecision": "deny", + "permissionDecisionReason": "Security policy violation", + "updatedInput": {"modified": "input"}, + }, + } + + transport = MockTransport() + hooks = { + "PreToolUse": [ + {"matcher": {"tool": "TestTool"}, "hooks": [comprehensive_hook]} + ] + } + + query = Query( + transport=transport, is_streaming_mode=True, can_use_tool=None, hooks=hooks + ) + + callback_id = "test_comprehensive_hook" + query.hook_callbacks[callback_id] = comprehensive_hook + + request = { + "type": "control_request", + "request_id": "test-comprehensive", + "request": { + "subtype": "hook_callback", + "callback_id": callback_id, + "input": {"test": "data"}, + "tool_use_id": "tool-456", + }, + } + + await query._handle_control_request(request) + + # Check response contains all the fields + assert len(transport.written_messages) > 0 + last_response = transport.written_messages[-1] + + # Parse the JSON response + response_data = json.loads(last_response) + # The hook result is nested at response.response + result = response_data["response"]["response"] + + # Verify control fields are present + assert result.get("continue_") is True or result.get("continue") is True + assert result.get("suppressOutput") is False + assert result.get("stopReason") == "Test stop reason" + + # Verify decision fields are present + assert result.get("decision") == "block" + assert result.get("reason") == "Test reason for blocking" + assert result.get("systemMessage") == "Test system message" + + # Verify hook-specific output is present + hook_output = result.get("hookSpecificOutput", {}) + assert hook_output.get("hookEventName") == "PreToolUse" + assert hook_output.get("permissionDecision") == "deny" + assert ( + hook_output.get("permissionDecisionReason") == "Security policy violation" + ) + assert "updatedInput" in hook_output + + @pytest.mark.asyncio + async def test_async_hook_output(self): + """Test AsyncHookJSONOutput type with proper async fields.""" + + async def async_hook( + input_data: dict, tool_use_id: str | None, context: HookContext + ) -> HookJSONOutput: + # Test that async hooks properly use async_ and asyncTimeout fields + return { + "async_": True, + "asyncTimeout": 5000, + } + + transport = MockTransport() + hooks = {"PreToolUse": [{"matcher": None, "hooks": [async_hook]}]} + + query = Query( + transport=transport, is_streaming_mode=True, can_use_tool=None, hooks=hooks + ) + + callback_id = "test_async_hook" + query.hook_callbacks[callback_id] = async_hook + + request = { + "type": "control_request", + "request_id": "test-async", + "request": { + "subtype": "hook_callback", + "callback_id": callback_id, + "input": {"test": "async_data"}, + "tool_use_id": None, + }, + } + + await query._handle_control_request(request) + + # Check response contains async fields + assert len(transport.written_messages) > 0 + last_response = transport.written_messages[-1] + + # Parse the JSON response + response_data = json.loads(last_response) + # The hook result is nested at response.response + result = response_data["response"]["response"] + + # The SDK should preserve the async_ field (or convert to "async") + assert result.get("async_") is True or result.get("async") is True + assert result.get("asyncTimeout") == 5000 + class TestClaudeAgentOptionsIntegration: """Test that callbacks work through ClaudeAgentOptions."""