feat: Add tool permission and hook callbacks support (#143)

km-anthropic · dicksontsai · web-flow · commit 68f0d7aa7dd9 · 2025-09-03T10:16:11.000-07:00
## Summary Adds comprehensive support for tool permission callbacks and hook callbacks to the Python SDK, enabling fine-grained control over tool execution and custom event handling. ## Key Changes - **Tool Permission Callbacks**: Control which tools Claude can use and modify their inputs - type with async support - with suggestions from CLI - for structured responses - **Hook Callbacks**: React to events in the Claude workflow - type for event handlers - for conditional hook execution - Support for tool_use_start, tool_use_end events - **Integration**: Full plumbing through ClaudeCodeOptions → Client → Query - **Examples**: Comprehensive example showing permission control patterns - **Tests**: Coverage for all callback scenarios ## Implementation Details - Callbacks are registered during initialization phase - Control protocol handles can_use_tool and hook_callback requests - Backwards compatible with dict returns for tool permissions - Proper error handling and type safety throughout Builds on top of #139's control protocol implementation. --------- Co-authored-by: Dickson Tsai <dickson@anthropic.com>
diff --git a/examples/tool_permission_callback.py b/examples/tool_permission_callback.py
@@ -0,0 +1,158 @@
+#!/usr/bin/env python3
+"""Example: Tool Permission Callbacks.
+
+This example demonstrates how to use tool permission callbacks to control
+which tools Claude can use and modify their inputs.
+"""
+
+import asyncio
+import json
+
+from claude_code_sdk import (
+    AssistantMessage,
+    ClaudeCodeOptions,
+    ClaudeSDKClient,
+    PermissionResultAllow,
+    PermissionResultDeny,
+    ResultMessage,
+    TextBlock,
+    ToolPermissionContext,
+)
+
+# Track tool usage for demonstration
+tool_usage_log = []
+
+
+async def my_permission_callback(
+    tool_name: str,
+    input_data: dict,
+    context: ToolPermissionContext
+) -> PermissionResultAllow | PermissionResultDeny:
+    """Control tool permissions based on tool type and input."""
+
+    # Log the tool request
+    tool_usage_log.append({
+        "tool": tool_name,
+        "input": input_data,
+        "suggestions": context.suggestions
+    })
+
+    print(f"\n🔧 Tool Permission Request: {tool_name}")
+    print(f"   Input: {json.dumps(input_data, indent=2)}")
+
+    # Always allow read operations
+    if tool_name in ["Read", "Glob", "Grep"]:
+        print(f"   ✅ Automatically allowing {tool_name} (read-only operation)")
+        return PermissionResultAllow()
+
+    # Deny write operations to system directories
+    if tool_name in ["Write", "Edit", "MultiEdit"]:
+        file_path = input_data.get("file_path", "")
+        if file_path.startswith("/etc/") or file_path.startswith("/usr/"):
+            print(f"   ❌ Denying write to system directory: {file_path}")
+            return PermissionResultDeny(
+                message=f"Cannot write to system directory: {file_path}"
+            )
+
+        # Redirect writes to a safe directory
+        if not file_path.startswith("/tmp/") and not file_path.startswith("./"):
+            safe_path = f"./safe_output/{file_path.split('/')[-1]}"
+            print(f"   ⚠️  Redirecting write from {file_path} to {safe_path}")
+            modified_input = input_data.copy()
+            modified_input["file_path"] = safe_path
+            return PermissionResultAllow(
+                updatedInput=modified_input
+            )
+
+    # Check dangerous bash commands
+    if tool_name == "Bash":
+        command = input_data.get("command", "")
+        dangerous_commands = ["rm -rf", "sudo", "chmod 777", "dd if=", "mkfs"]
+
+        for dangerous in dangerous_commands:
+            if dangerous in command:
+                print(f"   ❌ Denying dangerous command: {command}")
+                return PermissionResultDeny(
+                    message=f"Dangerous command pattern detected: {dangerous}"
+                )
+
+        # Allow but log the command
+        print(f"   ✅ Allowing bash command: {command}")
+        return PermissionResultAllow()
+
+    # For all other tools, ask the user
+    print(f"   ❓ Unknown tool: {tool_name}")
+    print(f"      Input: {json.dumps(input_data, indent=6)}")
+    user_input = input("   Allow this tool? (y/N): ").strip().lower()
+
+    if user_input in ("y", "yes"):
+        return PermissionResultAllow()
+    else:
+        return PermissionResultDeny(
+            message="User denied permission"
+        )
+
+
+async def main():
+    """Run example with tool permission callbacks."""
+
+    print("=" * 60)
+    print("Tool Permission Callback Example")
+    print("=" * 60)
+    print("\nThis example demonstrates how to:")
+    print("1. Allow/deny tools based on type")
+    print("2. Modify tool inputs for safety")
+    print("3. Log tool usage")
+    print("4. Prompt for unknown tools")
+    print("=" * 60)
+
+    # Configure options with our callback
+    options = ClaudeCodeOptions(
+        can_use_tool=my_permission_callback,
+        # Use default permission mode to ensure callbacks are invoked
+        permission_mode="default",
+        cwd="."  # Set working directory
+    )
+
+    # Create client and send a query that will use multiple tools
+    async with ClaudeSDKClient(options) as client:
+        print("\n📝 Sending query to Claude...")
+        await client.query(
+            "Please do the following:\n"
+            "1. List the files in the current directory\n"
+            "2. Create a simple Python hello world script at hello.py\n"
+            "3. Run the script to test it"
+        )
+
+        print("\n📨 Receiving response...")
+        message_count = 0
+
+        async for message in client.receive_response():
+            message_count += 1
+
+            if isinstance(message, AssistantMessage):
+                # Print Claude's text responses
+                for block in message.content:
+                    if isinstance(block, TextBlock):
+                        print(f"\n💬 Claude: {block.text}")
+
+            elif isinstance(message, ResultMessage):
+                print("\n✅ Task completed!")
+                print(f"   Duration: {message.duration_ms}ms")
+                if message.total_cost_usd:
+                    print(f"   Cost: ${message.total_cost_usd:.4f}")
+                print(f"   Messages processed: {message_count}")
+
+    # Print tool usage summary
+    print("\n" + "=" * 60)
+    print("Tool Usage Summary")
+    print("=" * 60)
+    for i, usage in enumerate(tool_usage_log, 1):
+        print(f"\n{i}. Tool: {usage['tool']}")
+        print(f"   Input: {json.dumps(usage['input'], indent=6)}")
+        if usage['suggestions']:
+            print(f"   Suggestions: {usage['suggestions']}")
+
+
+if __name__ == "__main__":
+    asyncio.run(main())
diff --git a/src/claude_code_sdk/__init__.py b/src/claude_code_sdk/__init__.py
@@ -16,16 +16,25 @@
 from .query import query
 from .types import (
     AssistantMessage,
+    CanUseTool,
     ClaudeCodeOptions,
     ContentBlock,
+    HookCallback,
+    HookContext,
+    HookMatcher,
     McpSdkServerConfig,
     McpServerConfig,
     Message,
     PermissionMode,
+    PermissionResult,
+    PermissionResultAllow,
+    PermissionResultDeny,
+    PermissionUpdate,
     ResultMessage,
     SystemMessage,
     TextBlock,
     ThinkingBlock,
+    ToolPermissionContext,
     ToolResultBlock,
     ToolUseBlock,
     UserMessage,
@@ -286,6 +295,16 @@ async def call_tool(name: str, arguments: dict) -> Any:
     "ToolUseBlock",
     "ToolResultBlock",
     "ContentBlock",
+    # Tool callbacks
+    "CanUseTool",
+    "ToolPermissionContext",
+    "PermissionResult",
+    "PermissionResultAllow",
+    "PermissionResultDeny",
+    "PermissionUpdate",
+    "HookCallback",
+    "HookContext",
+    "HookMatcher",
     # MCP Server Support
     "create_sdk_mcp_server",
     "tool",
diff --git a/src/claude_code_sdk/_internal/client.py b/src/claude_code_sdk/_internal/client.py
@@ -19,6 +19,22 @@ class InternalClient:
     def __init__(self) -> None:
         """Initialize the internal client."""
 
+    def _convert_hooks_to_internal_format(
+        self, hooks: dict[str, list]
+    ) -> dict[str, list[dict[str, Any]]]:
+        """Convert HookMatcher format to internal Query format."""
+        internal_hooks = {}
+        for event, matchers in hooks.items():
+            internal_hooks[event] = []
+            for matcher in matchers:
+                # Convert HookMatcher to internal dict format
+                internal_matcher = {
+                    "matcher": matcher.matcher if hasattr(matcher, 'matcher') else None,
+                    "hooks": matcher.hooks if hasattr(matcher, 'hooks') else []
+                }
+                internal_hooks[event].append(internal_matcher)
+        return internal_hooks
+
     async def process_query(
         self,
         prompt: str | AsyncIterable[dict[str, Any]],
@@ -48,8 +64,8 @@ async def process_query(
         query = Query(
             transport=chosen_transport,
             is_streaming_mode=is_streaming,
-            can_use_tool=None,  # TODO: Add support for can_use_tool callback
-            hooks=None,  # TODO: Add support for hooks
+            can_use_tool=options.can_use_tool,
+            hooks=self._convert_hooks_to_internal_format(options.hooks) if options.hooks else None,
             sdk_mcp_servers=sdk_mcp_servers,
         )
 
diff --git a/src/claude_code_sdk/_internal/query.py b/src/claude_code_sdk/_internal/query.py
@@ -15,10 +15,14 @@
 )
 
 from ..types import (
+    PermissionResult,
+    PermissionResultAllow,
+    PermissionResultDeny,
     SDKControlPermissionRequest,
     SDKControlRequest,
     SDKControlResponse,
     SDKHookCallbackRequest,
+    ToolPermissionContext,
 )
 from .transport import Transport
 
@@ -195,15 +199,34 @@ async def _handle_control_request(self, request: SDKControlRequest) -> None:
                 if not self.can_use_tool:
                     raise Exception("canUseTool callback is not provided")
 
-                response_data = await self.can_use_tool(
+                context = ToolPermissionContext(
+                    signal=None,  # TODO: Add abort signal support
+                    suggestions=permission_request.get("permission_suggestions", [])
+                )
+
+                response = await self.can_use_tool(
                     permission_request["tool_name"],
                     permission_request["input"],
-                    {
-                        "signal": None,  # TODO: Add abort signal support
-                        "suggestions": permission_request.get("permission_suggestions"),
-                    },
+                    context
                 )
 
+                # Convert PermissionResult to expected dict format
+                if isinstance(response, PermissionResultAllow):
+                    response_data = {
+                        "allow": True
+                    }
+                    if response.updatedInput is not None:
+                        response_data["input"] = response.updatedInput
+                    # TODO: Handle updatedPermissions when control protocol supports it
+                elif isinstance(response, PermissionResultDeny):
+                    response_data = {
+                        "allow": False,
+                        "reason": response.message
+                    }
+                    # TODO: Handle interrupt flag when control protocol supports it
+                else:
+                    raise TypeError(f"Tool permission callback must return PermissionResult (PermissionResultAllow or PermissionResultDeny), got {type(response)}")
+
             elif subtype == "hook_callback":
                 hook_callback_request: SDKHookCallbackRequest = request_data  # type: ignore[assignment]
                 # Handle hook callback
diff --git a/src/claude_code_sdk/client.py b/src/claude_code_sdk/client.py
@@ -100,6 +100,22 @@ def __init__(self, options: ClaudeCodeOptions | None = None):
         self._query: Any | None = None
         os.environ["CLAUDE_CODE_ENTRYPOINT"] = "sdk-py-client"
 
+    def _convert_hooks_to_internal_format(
+        self, hooks: dict[str, list]
+    ) -> dict[str, list[dict[str, Any]]]:
+        """Convert HookMatcher format to internal Query format."""
+        internal_hooks = {}
+        for event, matchers in hooks.items():
+            internal_hooks[event] = []
+            for matcher in matchers:
+                # Convert HookMatcher to internal dict format
+                internal_matcher = {
+                    "matcher": matcher.matcher if hasattr(matcher, 'matcher') else None,
+                    "hooks": matcher.hooks if hasattr(matcher, 'hooks') else []
+                }
+                internal_hooks[event].append(internal_matcher)
+        return internal_hooks
+
     async def connect(
         self, prompt: str | AsyncIterable[dict[str, Any]] | None = None
     ) -> None:
@@ -135,8 +151,8 @@ async def _empty_stream() -> AsyncIterator[dict[str, Any]]:
         self._query = Query(
             transport=self._transport,
             is_streaming_mode=True,  # ClaudeSDKClient always uses streaming mode
-            can_use_tool=None,  # TODO: Add support for can_use_tool callback
-            hooks=None,  # TODO: Add support for hooks
+            can_use_tool=self.options.can_use_tool,
+            hooks=self._convert_hooks_to_internal_format(self.options.hooks) if self.options.hooks else None,
             sdk_mcp_servers=sdk_mcp_servers,
         )
 
diff --git a/src/claude_code_sdk/types.py b/src/claude_code_sdk/types.py
diff --git a/tests/test_tool_callbacks.py b/tests/test_tool_callbacks.py
