mirror of
https://github.com/anthropics/claude-code-sdk-python.git
synced 2025-12-23 09:19:52 +00:00
de419792ef
This commit completes the canUseTool permission functionality implementation in the Python SDK to match the TypeScript SDK behavior: ## Key Changes: 1. **Transport Layer Enhancement** (`subprocess_cli.py`): - Added `--can-use-tool` CLI flag when canUseTool callback is provided - Ensures the CLI process knows to enable permission callback mode 2. **Example Implementation** (`can_use_tool_example.py`): - Comprehensive example showing how to use canUseTool callbacks - Demonstrates security policies, input modification, and denial patterns - Shows integration with ClaudeSDKClient in streaming mode ## Features Already Implemented: - **Type Definitions**: CanUseTool, PermissionResult*, ToolPermissionContext in `types.py` - **Client Integration**: Full canUseTool support in ClaudeSDKClient with proper validation - **Query Support**: Both query() function and ClaudeSDKClient handle canUseTool callbacks - **Control Protocol**: Complete bidirectional permission request/response handling - **Input Validation**: Ensures streaming mode is required, mutual exclusion with permission_prompt_tool_name - **Comprehensive Tests**: Full test coverage in test_tool_callbacks.py The implementation follows the TypeScript SDK's architecture exactly, providing: - Tool permission interception before execution - Input modification capabilities - Permission updates and suggestions - Proper error handling and validation 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude <noreply@anthropic.com>
90 lines
No EOL
2.9 KiB
Python
90 lines
No EOL
2.9 KiB
Python
#!/usr/bin/env python3
|
|
"""Example demonstrating canUseTool permission functionality."""
|
|
|
|
import asyncio
|
|
from claude_agent_sdk import (
|
|
ClaudeAgentOptions,
|
|
ClaudeSDKClient,
|
|
CanUseTool,
|
|
PermissionResultAllow,
|
|
PermissionResultDeny,
|
|
ToolPermissionContext,
|
|
)
|
|
|
|
|
|
async def security_callback(
|
|
tool_name: str, input_data: dict, context: ToolPermissionContext
|
|
) -> PermissionResultAllow | PermissionResultDeny:
|
|
"""Example security callback that implements tool permission policy."""
|
|
print(f"🔒 Permission check for tool: {tool_name}")
|
|
print(f" Input: {input_data}")
|
|
|
|
# Example policy: deny dangerous bash commands
|
|
if tool_name == "Bash":
|
|
command = input_data.get("command", "")
|
|
dangerous_patterns = ["rm -rf", "dd if=", "mkfs", "format", "del *"]
|
|
|
|
if any(pattern in command for pattern in dangerous_patterns):
|
|
print("❌ Denied: Command contains dangerous patterns")
|
|
return PermissionResultDeny(
|
|
message="Command blocked due to security policy",
|
|
interrupt=True
|
|
)
|
|
|
|
# Example: Allow but modify file operations to add safety
|
|
if tool_name in ["Write", "Edit"]:
|
|
file_path = input_data.get("file_path", "")
|
|
if file_path.startswith("/etc/") or file_path.startswith("/system/"):
|
|
print("⚠️ Modified: Adding backup for system file operation")
|
|
modified_input = input_data.copy()
|
|
modified_input["backup"] = True
|
|
return PermissionResultAllow(updated_input=modified_input)
|
|
|
|
print("✅ Allowed")
|
|
return PermissionResultAllow()
|
|
|
|
|
|
async def streaming_messages():
|
|
"""Generate streaming messages for testing."""
|
|
yield {
|
|
"type": "user",
|
|
"message": {"role": "user", "content": "What is 2+2?"},
|
|
"parent_tool_use_id": None,
|
|
"session_id": "test_session"
|
|
}
|
|
|
|
|
|
async def main():
|
|
"""Main example function."""
|
|
print("🚀 Testing canUseTool functionality")
|
|
|
|
try:
|
|
# Create options with canUseTool callback
|
|
options = ClaudeAgentOptions(
|
|
can_use_tool=security_callback,
|
|
permission_mode="default", # This will be overridden to use stdio
|
|
)
|
|
|
|
print("📋 Options created with canUseTool callback")
|
|
|
|
# Test with ClaudeSDKClient (streaming mode required for canUseTool)
|
|
client = ClaudeSDKClient(options=options)
|
|
|
|
print("🔌 Connecting to Claude with canUseTool enabled...")
|
|
await client.connect(streaming_messages())
|
|
|
|
print("✅ Connection successful!")
|
|
print("🔒 canUseTool callback is now active and will intercept tool usage")
|
|
|
|
# Note: In a real scenario, you would now interact with the client
|
|
# and the security_callback would be invoked for each tool use
|
|
|
|
await client.disconnect()
|
|
print("✅ Disconnected successfully")
|
|
|
|
except Exception as e:
|
|
print(f"❌ Error: {e}")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
asyncio.run(main()) |