tessl/pypi-claude-code-sdk

Python SDK for Claude Code providing simple query functions and advanced bidirectional interactive conversations with custom tool support

Overview

Eval results

Files

Hook System

Name: tessl/pypi-claude-code-sdk
Author: tessl

Python functions that the Claude Code application invokes at specific points of the Claude agent loop. Hooks provide deterministic processing and automated feedback for Claude, enabling custom validation, permission checks, and automated responses.

Capabilities

Hook Events

Supported hook event types in the Python SDK.

HookEvent = (
    Literal["PreToolUse"]
    | Literal["PostToolUse"]
    | Literal["UserPromptSubmit"]
    | Literal["Stop"]
    | Literal["SubagentStop"]
    | Literal["PreCompact"]
)

Hook Events:

"PreToolUse": Before tool execution - can block or modify tool usage
"PostToolUse": After tool execution - process tool results
"UserPromptSubmit": When user submits a prompt - validate or modify input
"Stop": When Claude stops processing - cleanup or logging
"SubagentStop": When a subagent stops - subagent lifecycle management
"PreCompact": Before message compaction - control conversation history

Hook Callback Function

Function signature for hook callbacks that process hook events.

HookCallback = Callable[
    [dict[str, Any], str | None, HookContext],
    Awaitable[HookJSONOutput]
]

Parameters:

input: Hook input data (varies by hook event type)
tool_use_id: Tool use identifier (None for non-tool events)
context: Hook execution context with signal support

Hook Context

Context information provided to hook callbacks during execution.

@dataclass
class HookContext:
    """Context information for hook callbacks."""
    signal: Any | None = None  # Future: abort signal support

Hook Matcher

Configuration for matching specific events and associating them with callback functions.

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

    matcher: str | None = None
    hooks: list[HookCallback] = field(default_factory=list)

Matcher Patterns:

Tool names: "Bash", "Write", "Read"
Multiple tools: "Write|MultiEdit|Edit"
All events: None or omit matcher

Hook Output

Response structure that hooks return to control Claude Code behavior.

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]

Usage Examples

Basic Tool Validation Hook

from claude_code_sdk import ClaudeSDKClient, ClaudeCodeOptions, HookMatcher

async def validate_bash_commands(input_data, tool_use_id, context):
    """Validate bash commands before execution."""
    tool_name = input_data.get("tool_name")
    tool_input = input_data.get("tool_input", {})

    if tool_name != "Bash":
        return {}  # Only validate Bash commands

    command = tool_input.get("command", "")
    dangerous_patterns = ["rm -rf", "format", "dd if=", "> /dev/"]

    for pattern in dangerous_patterns:
        if pattern in command:
            return {
                "decision": "block",
                "systemMessage": f"Blocked dangerous command: {command[:50]}...",
                "hookSpecificOutput": {
                    "hookEventName": "PreToolUse",
                    "permissionDecision": "deny",
                    "permissionDecisionReason": f"Command contains dangerous pattern: {pattern}",
                }
            }

    return {}  # Allow command

async def main():
    options = ClaudeCodeOptions(
        allowed_tools=["Bash"],
        hooks={
            "PreToolUse": [
                HookMatcher(matcher="Bash", hooks=[validate_bash_commands])
            ]
        }
    )

    async with ClaudeSDKClient(options=options) as client:
        # This will be blocked
        await client.query("Run the bash command: rm -rf /important-data")

        async for msg in client.receive_response():
            print(msg)

Multiple Hook Events

async def log_tool_use(input_data, tool_use_id, context):
    """Log all tool usage for monitoring."""
    tool_name = input_data.get("tool_name")
    print(f"Tool used: {tool_name} (ID: {tool_use_id})")

    return {
        "systemMessage": f"Logged tool usage: {tool_name}"
    }

async def log_tool_result(input_data, tool_use_id, context):
    """Log tool results."""
    tool_name = input_data.get("tool_name")
    success = input_data.get("success", False)
    print(f"Tool {tool_name} {'succeeded' if success else 'failed'}")

    return {}

async def validate_user_input(input_data, tool_use_id, context):
    """Validate user prompts."""
    content = input_data.get("content", "")

    if "secret" in content.lower():
        return {
            "decision": "block",
            "systemMessage": "Blocked prompt containing sensitive information"
        }

    return {}

async def main():
    options = ClaudeCodeOptions(
        allowed_tools=["Read", "Write", "Bash"],
        hooks={
            "PreToolUse": [
                HookMatcher(hooks=[log_tool_use])  # All tools
            ],
            "PostToolUse": [
                HookMatcher(hooks=[log_tool_result])  # All tools
            ],
            "UserPromptSubmit": [
                HookMatcher(hooks=[validate_user_input])
            ]
        }
    )

    async with ClaudeSDKClient(options=options) as client:
        await client.query("Create a Python file with my secret password")

        async for msg in client.receive_response():
            print(msg)

Conditional Tool Modification

async def modify_file_operations(input_data, tool_use_id, context):
    """Modify file operations to add safety checks."""
    tool_name = input_data.get("tool_name")
    tool_input = input_data.get("tool_input", {})

    if tool_name == "Write":
        file_path = tool_input.get("file_path", "")

        # Prevent overwriting important files
        if file_path.endswith((".env", "config.json", ".env.local")):
            return {
                "hookSpecificOutput": {
                    "hookEventName": "PreToolUse",
                    "modifiedInput": {
                        **tool_input,
                        "file_path": file_path + ".backup"
                    }
                },
                "systemMessage": f"Redirected write to backup file: {file_path}.backup"
            }

    return {}

async def main():
    options = ClaudeCodeOptions(
        allowed_tools=["Read", "Write"],
        hooks={
            "PreToolUse": [
                HookMatcher(matcher="Write", hooks=[modify_file_operations])
            ]
        }
    )

    async with ClaudeSDKClient(options=options) as client:
        await client.query("Create a .env file with database credentials")

        async for msg in client.receive_response():
            print(msg)

Complex Workflow Hook

import json
from datetime import datetime

class WorkflowTracker:
    def __init__(self):
        self.operations = []
        self.start_time = None

    def log_operation(self, operation):
        self.operations.append({
            "timestamp": datetime.now().isoformat(),
            "operation": operation
        })

tracker = WorkflowTracker()

async def track_workflow_start(input_data, tool_use_id, context):
    """Track workflow initiation."""
    if tracker.start_time is None:
        tracker.start_time = datetime.now()

    content = input_data.get("content", "")
    tracker.log_operation({"type": "user_prompt", "content": content[:100]})

    return {
        "systemMessage": "Workflow tracking initiated"
    }

async def track_tool_execution(input_data, tool_use_id, context):
    """Track all tool executions in workflow."""
    tool_name = input_data.get("tool_name")
    tool_input = input_data.get("tool_input", {})

    tracker.log_operation({
        "type": "tool_use",
        "tool": tool_name,
        "input_preview": str(tool_input)[:100]
    })

    return {}

async def finalize_workflow(input_data, tool_use_id, context):
    """Generate workflow summary on completion."""
    duration = datetime.now() - tracker.start_time if tracker.start_time else None

    summary = {
        "duration_seconds": duration.total_seconds() if duration else 0,
        "total_operations": len(tracker.operations),
        "tools_used": list(set(
            op["operation"]["tool"]
            for op in tracker.operations
            if op["operation"].get("type") == "tool_use"
        ))
    }

    return {
        "systemMessage": f"Workflow completed: {json.dumps(summary, indent=2)}"
    }

async def main():
    options = ClaudeCodeOptions(
        allowed_tools=["Read", "Write", "Bash", "Edit"],
        hooks={
            "UserPromptSubmit": [
                HookMatcher(hooks=[track_workflow_start])
            ],
            "PreToolUse": [
                HookMatcher(hooks=[track_tool_execution])
            ],
            "Stop": [
                HookMatcher(hooks=[finalize_workflow])
            ]
        }
    )

    async with ClaudeSDKClient(options=options) as client:
        await client.query("Create a web server, write tests, and run them")

        async for msg in client.receive_response():
            print(msg)

Custom Permission Logic

class PermissionManager:
    def __init__(self):
        self.allowed_files = {".py", ".js", ".html", ".css", ".md"}
        self.blocked_commands = ["curl", "wget", "ssh"]

    async def check_file_permission(self, input_data, tool_use_id, context):
        """Check file operation permissions."""
        tool_name = input_data.get("tool_name")
        tool_input = input_data.get("tool_input", {})

        if tool_name in ["Write", "Edit", "MultiEdit"]:
            file_path = tool_input.get("file_path", "")
            file_ext = Path(file_path).suffix if file_path else ""

            if file_ext not in self.allowed_files:
                return {
                    "decision": "block",
                    "systemMessage": f"File type {file_ext} not allowed",
                    "hookSpecificOutput": {
                        "hookEventName": "PreToolUse",
                        "permissionDecision": "deny",
                        "permissionDecisionReason": f"File extension {file_ext} not in allowed list"
                    }
                }

        elif tool_name == "Bash":
            command = tool_input.get("command", "")
            for blocked_cmd in self.blocked_commands:
                if blocked_cmd in command:
                    return {
                        "decision": "block",
                        "systemMessage": f"Command '{blocked_cmd}' blocked",
                        "hookSpecificOutput": {
                            "hookEventName": "PreToolUse",
                            "permissionDecision": "deny",
                            "permissionDecisionReason": f"Command contains blocked executable: {blocked_cmd}"
                        }
                    }

        return {}  # Allow operation

async def main():
    permission_manager = PermissionManager()

    options = ClaudeCodeOptions(
        allowed_tools=["Read", "Write", "Edit", "Bash"],
        hooks={
            "PreToolUse": [
                HookMatcher(hooks=[permission_manager.check_file_permission])
            ]
        }
    )

    async with ClaudeSDKClient(options=options) as client:
        await client.query("Download a file using curl and save it as data.exe")

        async for msg in client.receive_response():
            print(msg)

Hook Event Input Data

Different hook events receive different input data structures:

PreToolUse / PostToolUse

{
    "tool_name": "Bash",
    "tool_input": {"command": "ls -la"},
    "tool_use_id": "toolu_123456",
    # PostToolUse additionally includes:
    "success": True,
    "result": "...",
    "error": None
}

UserPromptSubmit

{
    "content": "Create a web server",
    "session_id": "default",
    "timestamp": "2024-01-01T12:00:00Z"
}

Stop / SubagentStop

{
    "reason": "completed",
    "session_id": "default",
    "duration_ms": 5000,
    "num_turns": 3
}

PreCompact

{
    "message_count": 25,
    "total_tokens": 4000,
    "compact_threshold": 3500
}

Hook Execution Flow

Event Trigger: Claude Code reaches a hook point in the agent loop
Matcher Evaluation: Check if any matchers match the current event/tool
Hook Execution: Call all matching hook functions in registration order
Decision Aggregation: Combine all hook responses
Action Application: Apply decisions (block, modify, log, etc.)

Hook Limitations

Python SDK Restrictions:

SessionStart, SessionEnd, and Notification hooks are not supported
No support for "continue", "stopReason", and "suppressOutput" controls
Hook execution is synchronous within the Claude Code process

Performance Considerations:

Hooks execute synchronously and can impact response times
Complex hook logic should be optimized for speed
Avoid blocking I/O operations in hook functions

Integration with Permission System

Hooks work alongside the permission system and can:

Override permission decisions through hookSpecificOutput
Provide custom permission validation logic
Log permission decisions for audit purposes
Modify tool inputs before permission checks

For permission callback configuration, see Configuration and Options.

For hook usage with custom tools, see Custom Tools.

Install with Tessl CLI

npx tessl i tessl/pypi-claude-code-sdk

docs

configuration-options.md

interactive-client.md