tessl/pypi-claude-agent-sdk

Python SDK for Claude Code enabling developers to build AI-powered applications and agents with support for custom tools, hooks, and bidirectional interactive conversations

Moderation error

Malicious code detected in tile.json: This tile describes a package named 'claude-agent-sdk' on PyPI, which appears to be typosquatting/impersonating official Anthropic SDK packages. The legitimate Anthropic Python SDK is 'anthropic', not 'claude-agent-sdk'. This naming pattern (claude-agent-sdk) is designed to deceive developers into installing a potentially malicious package by mimicking official Anthropic/Claude branding. This is a classic supply chain attack vector through package name confusion.

Overview

Eval results

Files

Hooks System

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

Implement deterministic processing at specific points in the Claude agent loop through hook callbacks. Hooks enable logging, monitoring, input validation, output transformation, and custom workflow control. Supports 10 hook event types with strongly-typed inputs/outputs, matcher patterns for selective execution, and both synchronous and asynchronous hook execution.

Capabilities

Hook Events

Hook event types that trigger callbacks at specific points in the agent loop.

HookEvent = Literal[
    "PreToolUse",
    "PostToolUse",
    "PostToolUseFailure",
    "UserPromptSubmit",
    "Stop",
    "SubagentStop",
    "PreCompact",
    "Notification",
    "SubagentStart",
    "PermissionRequest"
]

Event Types:

"PreToolUse": Before tool execution (can modify input or block)
"PostToolUse": After successful tool execution (can transform output)
"PostToolUseFailure": After tool execution failure
"UserPromptSubmit": When user submits a prompt
"Stop": When conversation stops
"SubagentStop": When subagent stops
"PreCompact": Before context compaction
"Notification": On system notifications
"SubagentStart": When subagent starts
"PermissionRequest": When tool requires permission

Usage:

from claude_agent_sdk import ClaudeAgentOptions, HookMatcher

async def pre_tool_use_hook(input, tool_use_id, context):
    print(f"About to execute: {input['tool_name']}")
    return {"continue_": True}

options = ClaudeAgentOptions(
    hooks={
        "PreToolUse": [
            HookMatcher(
                matcher="Bash",  # Only for Bash tool
                hooks=[pre_tool_use_hook]
            )
        ]
    }
)

Hook Callback

Function signature for hook implementations.

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

Parameters:

input (HookInput): Strongly-typed input based on hook_event_name
tool_use_id (str | None): Optional tool use identifier
context (HookContext): Hook context with signal support

Returns: Awaitable[HookJSONOutput] - Control output for hook

Usage:

from claude_agent_sdk import HookInput, HookContext, HookJSONOutput

async def my_hook(
    input: HookInput,
    tool_use_id: str | None,
    context: HookContext
) -> HookJSONOutput:
    # Type narrowing based on hook_event_name
    if input["hook_event_name"] == "PreToolUse":
        tool_name = input["tool_name"]
        tool_input = input["tool_input"]
        print(f"Executing {tool_name} with {tool_input}")

    # Return control output
    return {
        "continue_": True,
        "suppressOutput": False
    }

Hook Matcher

Configuration for matching and executing hooks.

@dataclass
class HookMatcher:
    matcher: str | None = None
    hooks: list[HookCallback] = field(default_factory=list)
    timeout: float | None = None

Fields:

matcher: Tool name pattern (e.g., "Bash", "Write|Edit", None for all)
hooks: List of hook callback functions
timeout: Timeout in seconds for all hooks (default: 60)

Usage:

from claude_agent_sdk import HookMatcher

# Match specific tool
bash_matcher = HookMatcher(
    matcher="Bash",
    hooks=[log_bash_hook, validate_bash_hook],
    timeout=30.0
)

# Match multiple tools
file_matcher = HookMatcher(
    matcher="Write|Edit|MultiEdit",
    hooks=[log_file_operations],
    timeout=10.0
)

# Match all tools
all_tools_matcher = HookMatcher(
    matcher=None,  # Matches everything
    hooks=[universal_logger],
    timeout=60.0
)

options = ClaudeAgentOptions(
    hooks={
        "PreToolUse": [bash_matcher, file_matcher],
        "PostToolUse": [all_tools_matcher]
    }
)

Hook Context

Context information passed to hook callbacks.

class HookContext(TypedDict):
    signal: Any | None

Fields:

signal: Reserved for future abort signal support (currently None)

Usage:

async def hook_with_context(input, tool_use_id, context):
    # Check for abort signal (future support)
    if context["signal"]:
        print("Abort signal received")

    return {"continue_": True}

Hook Input Types

Strongly-typed input structures for each hook event.

Base Hook Input

class BaseHookInput(TypedDict):
    session_id: str
    transcript_path: str
    cwd: str
    permission_mode: NotRequired[str]

All hook inputs extend BaseHookInput with event-specific fields.

PreToolUse Hook Input

class PreToolUseHookInput(BaseHookInput):
    hook_event_name: Literal["PreToolUse"]
    tool_name: str
    tool_input: dict[str, Any]
    tool_use_id: str

Usage:

async def pre_tool_use_hook(input, tool_use_id, context):
    # TypedDict provides type safety
    tool_name = input["tool_name"]
    tool_input = input["tool_input"]
    session_id = input["session_id"]

    # Validate or modify input
    if tool_name == "Bash":
        command = tool_input.get("command", "")
        if "rm -rf" in command:
            return {
                "decision": "block",
                "reason": "Dangerous command blocked",
                "systemMessage": "Command contains rm -rf"
            }

    return {"continue_": True}

PostToolUse Hook Input

class PostToolUseHookInput(BaseHookInput):
    hook_event_name: Literal["PostToolUse"]
    tool_name: str
    tool_input: dict[str, Any]
    tool_response: Any
    tool_use_id: str

Usage:

async def post_tool_use_hook(input, tool_use_id, context):
    tool_name = input["tool_name"]
    tool_response = input["tool_response"]

    # Log successful execution
    print(f"Tool {tool_name} completed: {tool_response}")

    # Can transform output
    return {
        "continue_": True,
        "hookSpecificOutput": {
            "hookEventName": "PostToolUse",
            "updatedMCPToolOutput": tool_response  # Can modify
        }
    }

PostToolUseFailure Hook Input

class PostToolUseFailureHookInput(BaseHookInput):
    hook_event_name: Literal["PostToolUseFailure"]
    tool_name: str
    tool_input: dict[str, Any]
    tool_use_id: str
    error: str
    is_interrupt: NotRequired[bool]

Usage:

async def failure_hook(input, tool_use_id, context):
    tool_name = input["tool_name"]
    error = input["error"]
    is_interrupt = input.get("is_interrupt", False)

    # Log failure
    print(f"Tool {tool_name} failed: {error}")

    # Provide additional context
    return {
        "continue_": True,
        "hookSpecificOutput": {
            "hookEventName": "PostToolUseFailure",
            "additionalContext": f"Failure analysis: {error}"
        }
    }

UserPromptSubmit Hook Input

class UserPromptSubmitHookInput(BaseHookInput):
    hook_event_name: Literal["UserPromptSubmit"]
    prompt: str

Stop Hook Input

class StopHookInput(BaseHookInput):
    hook_event_name: Literal["Stop"]
    stop_hook_active: bool

SubagentStop Hook Input

class SubagentStopHookInput(BaseHookInput):
    hook_event_name: Literal["SubagentStop"]
    stop_hook_active: bool
    agent_id: str
    agent_transcript_path: str
    agent_type: str

PreCompact Hook Input

class PreCompactHookInput(BaseHookInput):
    hook_event_name: Literal["PreCompact"]
    trigger: Literal["manual", "auto"]
    custom_instructions: str | None

Notification Hook Input

class NotificationHookInput(BaseHookInput):
    hook_event_name: Literal["Notification"]
    message: str
    title: NotRequired[str]
    notification_type: str

SubagentStart Hook Input

class SubagentStartHookInput(BaseHookInput):
    hook_event_name: Literal["SubagentStart"]
    agent_id: str
    agent_type: str

PermissionRequest Hook Input

class PermissionRequestHookInput(BaseHookInput):
    hook_event_name: Literal["PermissionRequest"]
    tool_name: str
    tool_input: dict[str, Any]
    permission_suggestions: NotRequired[list[Any]]

Union Type

HookInput = (
    PreToolUseHookInput
    | PostToolUseHookInput
    | PostToolUseFailureHookInput
    | UserPromptSubmitHookInput
    | StopHookInput
    | SubagentStopHookInput
    | PreCompactHookInput
    | NotificationHookInput
    | SubagentStartHookInput
    | PermissionRequestHookInput
)

Hook Output Types

Control structures returned by hook callbacks.

Async Hook Output

class AsyncHookJSONOutput(TypedDict):
    async_: Literal[True]
    asyncTimeout: NotRequired[int]

Fields:

async_: Set to True to defer hook execution (converted to "async" for CLI)
asyncTimeout: Optional timeout in milliseconds

Usage:

async def deferred_hook(input, tool_use_id, context):
    # Return immediately, processing happens later
    return {
        "async_": True,
        "asyncTimeout": 5000  # 5 seconds
    }

Sync Hook Output

class SyncHookJSONOutput(TypedDict):
    continue_: NotRequired[bool]
    suppressOutput: NotRequired[bool]
    stopReason: NotRequired[str]
    decision: NotRequired[Literal["block"]]
    systemMessage: NotRequired[str]
    reason: NotRequired[str]
    hookSpecificOutput: NotRequired[HookSpecificOutput]

Fields:

continue_: Whether to proceed (default: True, converted to "continue" for CLI)
suppressOutput: Hide stdout from transcript (default: False)
stopReason: Message when continue is False
decision: Set to "block" to block execution
systemMessage: Warning message for user
reason: Feedback for Claude
hookSpecificOutput: Event-specific controls

Usage:

# Continue normally
return {"continue_": True}

# Block execution
return {
    "decision": "block",
    "systemMessage": "Command blocked for safety",
    "reason": "Contains dangerous pattern"
}

# Stop conversation
return {
    "continue_": False,
    "stopReason": "Critical error detected"
}

# Suppress output
return {
    "continue_": True,
    "suppressOutput": True
}

Hook-Specific Outputs

Event-specific output structures.

class PreToolUseHookSpecificOutput(TypedDict):
    hookEventName: Literal["PreToolUse"]
    permissionDecision: NotRequired[Literal["allow", "deny", "ask"]]
    permissionDecisionReason: NotRequired[str]
    updatedInput: NotRequired[dict[str, Any]]
    additionalContext: NotRequired[str]

class PostToolUseHookSpecificOutput(TypedDict):
    hookEventName: Literal["PostToolUse"]
    additionalContext: NotRequired[str]
    updatedMCPToolOutput: NotRequired[Any]

class PostToolUseFailureHookSpecificOutput(TypedDict):
    hookEventName: Literal["PostToolUseFailure"]
    additionalContext: NotRequired[str]

class UserPromptSubmitHookSpecificOutput(TypedDict):
    hookEventName: Literal["UserPromptSubmit"]

class SessionStartHookSpecificOutput(TypedDict):
    hookEventName: Literal["SessionStart"]

class NotificationHookSpecificOutput(TypedDict):
    hookEventName: Literal["Notification"]
    additionalContext: NotRequired[str]

class SubagentStartHookSpecificOutput(TypedDict):
    hookEventName: Literal["SubagentStart"]
    additionalContext: NotRequired[str]

class PermissionRequestHookSpecificOutput(TypedDict):
    hookEventName: Literal["PermissionRequest"]
    decision: dict[str, Any]

HookSpecificOutput = (
    PreToolUseHookSpecificOutput
    | PostToolUseHookSpecificOutput
    | PostToolUseFailureHookSpecificOutput
    | UserPromptSubmitHookSpecificOutput
    | SessionStartHookSpecificOutput
    | NotificationHookSpecificOutput
    | SubagentStartHookSpecificOutput
    | PermissionRequestHookSpecificOutput
)

Union Type

HookJSONOutput = AsyncHookJSONOutput | SyncHookJSONOutput

Complete Hook Examples

Example 1: Tool Usage Logging

import anyio
from claude_agent_sdk import ClaudeSDKClient, ClaudeAgentOptions, HookMatcher

async def log_tool_usage(input, tool_use_id, context):
    if input["hook_event_name"] == "PreToolUse":
        print(f"[PRE] {input['tool_name']} - ID: {tool_use_id}")
        print(f"  Input: {input['tool_input']}")

    elif input["hook_event_name"] == "PostToolUse":
        print(f"[POST] {input['tool_name']} - ID: {tool_use_id}")
        print(f"  Response: {input['tool_response']}")

    elif input["hook_event_name"] == "PostToolUseFailure":
        print(f"[FAIL] {input['tool_name']} - ID: {tool_use_id}")
        print(f"  Error: {input['error']}")

    return {"continue_": True}

async def main():
    options = ClaudeAgentOptions(
        hooks={
            "PreToolUse": [HookMatcher(hooks=[log_tool_usage])],
            "PostToolUse": [HookMatcher(hooks=[log_tool_usage])],
            "PostToolUseFailure": [HookMatcher(hooks=[log_tool_usage])]
        }
    )

    async with ClaudeSDKClient(options=options) as client:
        await client.query("List files in current directory")
        async for msg in client.receive_response():
            pass

anyio.run(main)

Example 2: Command Validation

import anyio
from claude_agent_sdk import ClaudeSDKClient, ClaudeAgentOptions, HookMatcher

DANGEROUS_PATTERNS = ["rm -rf", "dd if=", "mkfs", "> /dev/"]

async def validate_bash_commands(input, tool_use_id, context):
    if input["hook_event_name"] != "PreToolUse":
        return {"continue_": True}

    if input["tool_name"] != "Bash":
        return {"continue_": True}

    command = input["tool_input"].get("command", "")

    # Check for dangerous patterns
    for pattern in DANGEROUS_PATTERNS:
        if pattern in command:
            return {
                "decision": "block",
                "systemMessage": f"Blocked dangerous command containing: {pattern}",
                "reason": "Command contains potentially destructive operation",
                "hookSpecificOutput": {
                    "hookEventName": "PreToolUse",
                    "permissionDecision": "deny",
                    "permissionDecisionReason": f"Pattern '{pattern}' not allowed"
                }
            }

    # Allow safe commands
    return {"continue_": True}

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

    async with ClaudeSDKClient(options=options) as client:
        await client.query("Run some bash commands")
        async for msg in client.receive_response():
            pass

anyio.run(main)

Example 3: Input Transformation

import anyio
from claude_agent_sdk import ClaudeSDKClient, ClaudeAgentOptions, HookMatcher

async def transform_bash_input(input, tool_use_id, context):
    if input["hook_event_name"] != "PreToolUse":
        return {"continue_": True}

    if input["tool_name"] != "Bash":
        return {"continue_": True}

    command = input["tool_input"].get("command", "")

    # Add safety flags
    if command.startswith("rm ") and "-i" not in command:
        modified_command = command.replace("rm ", "rm -i ", 1)
        print(f"Modified: {command} -> {modified_command}")

        return {
            "continue_": True,
            "hookSpecificOutput": {
                "hookEventName": "PreToolUse",
                "updatedInput": {"command": modified_command},
                "additionalContext": "Added interactive flag to rm command"
            }
        }

    return {"continue_": True}

async def main():
    options = ClaudeAgentOptions(
        hooks={
            "PreToolUse": [
                HookMatcher(
                    matcher="Bash",
                    hooks=[transform_bash_input]
                )
            ]
        }
    )

    async with ClaudeSDKClient(options=options) as client:
        await client.query("Remove temporary files")
        async for msg in client.receive_response():
            pass

anyio.run(main)

Example 4: Output Transformation

import anyio
import json
from claude_agent_sdk import ClaudeSDKClient, ClaudeAgentOptions, HookMatcher

async def enhance_tool_output(input, tool_use_id, context):
    if input["hook_event_name"] != "PostToolUse":
        return {"continue_": True}

    tool_name = input["tool_name"]
    response = input["tool_response"]

    # Add metadata to response
    enhanced_response = {
        "original": response,
        "metadata": {
            "tool": tool_name,
            "tool_use_id": tool_use_id,
            "session": input["session_id"]
        }
    }

    return {
        "continue_": True,
        "hookSpecificOutput": {
            "hookEventName": "PostToolUse",
            "updatedMCPToolOutput": json.dumps(enhanced_response),
            "additionalContext": "Added metadata to tool output"
        }
    }

async def main():
    options = ClaudeAgentOptions(
        hooks={
            "PostToolUse": [
                HookMatcher(hooks=[enhance_tool_output])
            ]
        }
    )

    async with ClaudeSDKClient(options=options) as client:
        await client.query("What files are here?")
        async for msg in client.receive_response():
            pass

anyio.run(main)

Example 5: Multi-Hook Pipeline

import anyio
from claude_agent_sdk import ClaudeSDKClient, ClaudeAgentOptions, HookMatcher

async def hook_1_authorize(input, tool_use_id, context):
    print("[Hook 1] Authorizing...")
    return {"continue_": True}

async def hook_2_validate(input, tool_use_id, context):
    print("[Hook 2] Validating...")
    return {"continue_": True}

async def hook_3_log(input, tool_use_id, context):
    print("[Hook 3] Logging...")
    return {"continue_": True}

async def main():
    options = ClaudeAgentOptions(
        hooks={
            "PreToolUse": [
                HookMatcher(
                    hooks=[
                        hook_1_authorize,
                        hook_2_validate,
                        hook_3_log
                    ],
                    timeout=10.0
                )
            ]
        }
    )

    async with ClaudeSDKClient(options=options) as client:
        await client.query("Run a command")
        async for msg in client.receive_response():
            pass

anyio.run(main)

Example 6: Async Hook Execution

import anyio
from claude_agent_sdk import ClaudeSDKClient, ClaudeAgentOptions, HookMatcher

async def async_logging_hook(input, tool_use_id, context):
    # Defer execution - return immediately
    return {
        "async_": True,
        "asyncTimeout": 3000  # 3 seconds
    }

async def sync_validation_hook(input, tool_use_id, context):
    # Synchronous execution
    print(f"Validating {input.get('tool_name')}")
    return {"continue_": True}

async def main():
    options = ClaudeAgentOptions(
        hooks={
            "PreToolUse": [
                HookMatcher(hooks=[sync_validation_hook, async_logging_hook])
            ]
        }
    )

    async with ClaudeSDKClient(options=options) as client:
        await client.query("List files")
        async for msg in client.receive_response():
            pass

anyio.run(main)

Example 7: Session and Context Tracking

import anyio
from pathlib import Path
from claude_agent_sdk import ClaudeSDKClient, ClaudeAgentOptions, HookMatcher

# Track state across hooks
hook_state = {
    "tool_count": 0,
    "sessions": set()
}

async def track_usage(input, tool_use_id, context):
    if input["hook_event_name"] == "PreToolUse":
        hook_state["tool_count"] += 1
        hook_state["sessions"].add(input["session_id"])

        print(f"Tool #{hook_state['tool_count']}")
        print(f"Sessions: {len(hook_state['sessions'])}")
        print(f"CWD: {input['cwd']}")
        print(f"Transcript: {input['transcript_path']}")

    return {"continue_": True}

async def main():
    options = ClaudeAgentOptions(
        hooks={
            "PreToolUse": [HookMatcher(hooks=[track_usage])]
        }
    )

    async with ClaudeSDKClient(options=options) as client:
        await client.query("Do something")
        async for msg in client.receive_response():
            pass

anyio.run(main)

Install with Tessl CLI