CtrlK
BlogDocsLog inGet started
Tessl Logo

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

errors.mddocs/reference/

Error Handling

Comprehensive exception hierarchy for error handling with specific error types for different failure modes. All SDK errors derive from ClaudeSDKError, enabling catch-all error handling or specific error type handling.

Capabilities

Base Error

Base exception class for all SDK errors.

class ClaudeSDKError(Exception):
    """Base exception for all Claude SDK errors."""

Usage:

from claude_agent_sdk import query, ClaudeSDKError

try:
    async for message in query(prompt="Hello"):
        print(message)
except ClaudeSDKError as e:
    # Catches all SDK-related errors
    print(f"SDK error: {e}")

CLI Connection Error

Raised when unable to connect to Claude Code CLI.

class CLIConnectionError(ClaudeSDKError):
    """Raised when unable to connect to Claude Code."""

Common Causes:

  • Claude Code not installed
  • Claude Code process failed to start
  • Connection lost during operation
  • Invalid transport configuration

Usage:

from claude_agent_sdk import ClaudeSDKClient, CLIConnectionError

try:
    async with ClaudeSDKClient() as client:
        await client.query("Hello")
        async for msg in client.receive_response():
            print(msg)
except CLIConnectionError as e:
    print(f"Cannot connect to Claude Code: {e}")
    print("Make sure Claude Code is installed and accessible")

CLI Not Found Error

Raised when Claude Code CLI is not found or not installed.

class CLINotFoundError(CLIConnectionError):
    """Raised when Claude Code is not found or not installed."""

    def __init__(
        self,
        message: str = "Claude Code not found",
        cli_path: str | None = None
    )

Attributes:

  • message: Error message
  • cli_path: Path where CLI was expected (if specified)

Common Causes:

  • Claude Code not installed
  • CLI not in PATH
  • Invalid cli_path in options
  • Permission issues accessing CLI

Usage:

from claude_agent_sdk import query, CLINotFoundError, ClaudeAgentOptions

try:
    options = ClaudeAgentOptions(cli_path="/custom/path/to/claude-code")
    async for message in query(prompt="Hello", options=options):
        print(message)
except CLINotFoundError as e:
    print(f"CLI not found: {e}")
    print("Install Claude Code or check cli_path configuration")

Process Error

Raised when the CLI process fails during execution.

class ProcessError(ClaudeSDKError):
    """Raised when the CLI process fails."""

    def __init__(
        self,
        message: str,
        exit_code: int | None = None,
        stderr: str | None = None
    )

    exit_code: int | None
    stderr: str | None

Attributes:

  • message: Error message
  • exit_code: Process exit code (if available)
  • stderr: Error output from process (if available)

Common Causes:

  • CLI crashed or exited unexpectedly
  • Invalid CLI arguments
  • Insufficient system resources
  • Permission denied
  • Configuration error in CLI

Usage:

from claude_agent_sdk import query, ProcessError

try:
    async for message in query(prompt="Hello"):
        print(message)
except ProcessError as e:
    print(f"Process failed: {e}")
    if e.exit_code:
        print(f"Exit code: {e.exit_code}")
    if e.stderr:
        print(f"Error output: {e.stderr}")

CLI JSON Decode Error

Raised when unable to decode JSON from CLI output.

class CLIJSONDecodeError(ClaudeSDKError):
    """Raised when unable to decode JSON from CLI output."""

    def __init__(self, line: str, original_error: Exception)

    line: str
    original_error: Exception

Attributes:

  • line: The malformed JSON line (truncated to 100 chars)
  • original_error: The underlying JSON decode exception

Common Causes:

  • CLI output corruption
  • Version mismatch between SDK and CLI
  • CLI writing non-JSON to stdout
  • Truncated or incomplete JSON output

Usage:

from claude_agent_sdk import query, CLIJSONDecodeError

try:
    async for message in query(prompt="Hello"):
        print(message)
except CLIJSONDecodeError as e:
    print(f"Invalid JSON from CLI: {e}")
    print(f"Line: {e.line}")
    print(f"Original error: {e.original_error}")

Message Parse Error

Raised when unable to parse a message from CLI output.

Note: This error is internal and not included in __all__ exports. It's available for advanced error handling but not part of the primary public API.

class MessageParseError(ClaudeSDKError):
    """Raised when unable to parse a message from CLI output."""

    def __init__(
        self,
        message: str,
        data: dict[str, Any] | None = None
    )

    data: dict[str, Any] | None

Attributes:

  • message: Error description
  • data: The raw message data that failed to parse (if available)

Common Causes:

  • Unexpected message format from CLI
  • Missing required fields in message
  • Type mismatch in message data
  • Protocol version mismatch

Usage:

from claude_agent_sdk import query, MessageParseError

try:
    async for message in query(prompt="Hello"):
        print(message)
except MessageParseError as e:
    print(f"Cannot parse message: {e}")
    if e.data:
        print(f"Raw data: {e.data}")

Complete Error Handling Examples

Example 1: Basic Error Handling

import anyio
from claude_agent_sdk import query, ClaudeSDKError

async def main():
    try:
        async for message in query(prompt="What is 2 + 2?"):
            print(message)
    except ClaudeSDKError as e:
        print(f"Error: {e}")

anyio.run(main)

Example 2: Specific Error Handling

import anyio
from claude_agent_sdk import (
    query,
    CLINotFoundError,
    CLIConnectionError,
    ProcessError,
    CLIJSONDecodeError,
    ClaudeAgentOptions
)

async def main():
    try:
        options = ClaudeAgentOptions(
            cli_path="/usr/local/bin/claude-code"
        )
        async for message in query(prompt="Hello", options=options):
            print(message)

    except CLINotFoundError as e:
        print(f"Claude Code not installed: {e}")
        print("Run: pip install claude-code-cli")

    except ProcessError as e:
        print(f"Process crashed: {e}")
        if e.exit_code:
            print(f"Exit code: {e.exit_code}")
        if e.stderr:
            print(f"Stderr: {e.stderr}")

    except CLIJSONDecodeError as e:
        print(f"Invalid JSON output: {e}")
        print("This might indicate a version mismatch")

    except CLIConnectionError as e:
        print(f"Connection failed: {e}")
        print("Check if Claude Code is accessible")

anyio.run(main)

Example 3: Retry Logic

import anyio
from claude_agent_sdk import (
    query,
    CLIConnectionError,
    ProcessError,
    ClaudeSDKError
)

async def query_with_retry(prompt: str, max_retries: int = 3):
    for attempt in range(max_retries):
        try:
            async for message in query(prompt=prompt):
                yield message
            return  # Success

        except (CLIConnectionError, ProcessError) as e:
            if attempt < max_retries - 1:
                print(f"Attempt {attempt + 1} failed: {e}")
                print(f"Retrying... ({max_retries - attempt - 1} attempts left)")
                await anyio.sleep(2 ** attempt)  # Exponential backoff
            else:
                print(f"All {max_retries} attempts failed")
                raise

        except ClaudeSDKError:
            # Don't retry for other errors
            raise

async def main():
    try:
        async for message in query_with_retry("What is 2 + 2?"):
            print(message)
    except ClaudeSDKError as e:
        print(f"Final error: {e}")

anyio.run(main)

Example 4: Graceful Degradation

import anyio
from claude_agent_sdk import (
    query,
    ClaudeAgentOptions,
    CLINotFoundError,
    ProcessError
)

async def query_with_fallback(prompt: str):
    # Try with primary model
    try:
        options = ClaudeAgentOptions(model="opus")
        async for message in query(prompt=prompt, options=options):
            yield message
        return

    except ProcessError as e:
        print(f"Primary model failed: {e}")
        print("Trying fallback model...")

    # Try with fallback model
    try:
        options = ClaudeAgentOptions(
            model="sonnet",
            fallback_model="haiku"
        )
        async for message in query(prompt=prompt, options=options):
            yield message
        return

    except ProcessError as e:
        print(f"Fallback also failed: {e}")
        raise

async def main():
    try:
        async for message in query_with_fallback("Hello"):
            print(message)
    except Exception as e:
        print(f"All options exhausted: {e}")

anyio.run(main)

Example 5: Error Logging

import anyio
import logging
from claude_agent_sdk import (
    ClaudeSDKClient,
    ClaudeSDKError,
    CLIConnectionError,
    ProcessError,
    CLIJSONDecodeError
)

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

async def main():
    try:
        async with ClaudeSDKClient() as client:
            logger.info("Connected to Claude Code")

            await client.query("List files")
            async for msg in client.receive_response():
                logger.debug(f"Received: {msg}")

    except CLIConnectionError as e:
        logger.error(f"Connection error: {e}", exc_info=True)

    except ProcessError as e:
        logger.error(
            f"Process error: {e} (exit code: {e.exit_code})",
            exc_info=True
        )

    except CLIJSONDecodeError as e:
        logger.error(
            f"JSON decode error: {e}",
            extra={"line": e.line},
            exc_info=True
        )

    except ClaudeSDKError as e:
        logger.error(f"SDK error: {e}", exc_info=True)

anyio.run(main)

Example 6: Context Manager Error Handling

import anyio
from claude_agent_sdk import ClaudeSDKClient, ClaudeSDKError

async def main():
    client = ClaudeSDKClient()

    try:
        await client.connect()
        await client.query("Hello")

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

    except ClaudeSDKError as e:
        print(f"Error during conversation: {e}")

    finally:
        # Ensure cleanup even on error
        try:
            await client.disconnect()
        except Exception as e:
            print(f"Error during cleanup: {e}")

anyio.run(main)

Example 7: Custom Error Handler

import anyio
from claude_agent_sdk import (
    query,
    ClaudeSDKError,
    CLINotFoundError,
    CLIConnectionError,
    ProcessError,
    CLIJSONDecodeError,
    MessageParseError
)

class ErrorHandler:
    def __init__(self):
        self.errors = []

    def handle_error(self, error: Exception) -> bool:
        """Handle error and return True if should retry."""
        self.errors.append(error)

        if isinstance(error, CLINotFoundError):
            print("FATAL: Claude Code not found")
            return False

        elif isinstance(error, ProcessError):
            if error.exit_code == 137:  # OOM killed
                print("Process killed (likely out of memory)")
                return False
            print(f"Process error (code {error.exit_code}), retrying...")
            return True

        elif isinstance(error, CLIJSONDecodeError):
            print("JSON decode error, might be transient")
            return True

        elif isinstance(error, MessageParseError):
            print("Message parse error")
            return False

        elif isinstance(error, CLIConnectionError):
            print("Connection error, retrying...")
            return True

        return False

async def main():
    handler = ErrorHandler()

    for attempt in range(3):
        try:
            async for message in query(prompt="Hello"):
                print(message)
            break  # Success

        except ClaudeSDKError as e:
            should_retry = handler.handle_error(e)
            if not should_retry or attempt == 2:
                print(f"Giving up after {len(handler.errors)} errors")
                raise

    print(f"Succeeded after {len(handler.errors)} errors")

anyio.run(main)

Example 8: Error Recovery with State

import anyio
from claude_agent_sdk import (
    ClaudeSDKClient,
    ClaudeAgentOptions,
    CLIConnectionError,
    ResultMessage
)

async def main():
    # Save conversation state for recovery
    messages = []
    session_id = None

    try:
        async with ClaudeSDKClient() as client:
            await client.query("Start task")

            async for msg in client.receive_response():
                messages.append(msg)
                if isinstance(msg, ResultMessage):
                    session_id = msg.session_id

    except CLIConnectionError as e:
        print(f"Connection lost: {e}")

        if session_id:
            print(f"Attempting to resume session: {session_id}")

            try:
                # Resume from saved session
                options = ClaudeAgentOptions(
                    resume=session_id,
                    continue_conversation=True
                )
                async with ClaudeSDKClient(options=options) as client:
                    await client.query("Continue task")
                    async for msg in client.receive_response():
                        print(msg)

            except Exception as e:
                print(f"Recovery failed: {e}")

anyio.run(main)

Example 9: Validation and Error Prevention

import anyio
from pathlib import Path
from claude_agent_sdk import (
    query,
    ClaudeAgentOptions,
    CLINotFoundError,
    ClaudeSDKError
)

async def validate_options(options: ClaudeAgentOptions) -> None:
    """Validate options before query to catch errors early."""
    # Check CLI path if specified
    if options.cli_path:
        cli_path = Path(options.cli_path)
        if not cli_path.exists():
            raise CLINotFoundError(
                "Specified CLI path does not exist",
                cli_path=str(cli_path)
            )
        if not cli_path.is_file():
            raise ValueError(f"CLI path is not a file: {cli_path}")

    # Check working directory
    if options.cwd:
        cwd = Path(options.cwd)
        if not cwd.exists():
            raise ValueError(f"Working directory does not exist: {cwd}")
        if not cwd.is_dir():
            raise ValueError(f"CWD is not a directory: {cwd}")

    # Validate budget
    if options.max_budget_usd is not None:
        if options.max_budget_usd <= 0:
            raise ValueError("max_budget_usd must be positive")

async def main():
    options = ClaudeAgentOptions(
        cli_path="/usr/local/bin/claude-code",
        cwd="/workspace",
        max_budget_usd=10.0
    )

    try:
        # Validate before attempting query
        await validate_options(options)

        async for message in query(prompt="Hello", options=options):
            print(message)

    except ValueError as e:
        print(f"Validation error: {e}")

    except ClaudeSDKError as e:
        print(f"SDK error: {e}")

anyio.run(main)

Example 10: Comprehensive Error Reporting

import anyio
import traceback
from datetime import datetime
from claude_agent_sdk import (
    query,
    ClaudeAgentOptions,
    ClaudeSDKError
)

class ErrorReport:
    def __init__(self):
        self.timestamp = datetime.now()
        self.error_type = None
        self.error_message = None
        self.traceback = None
        self.context = {}

    def capture(self, error: Exception, context: dict = None):
        self.error_type = type(error).__name__
        self.error_message = str(error)
        self.traceback = traceback.format_exc()
        self.context = context or {}

    def __str__(self):
        lines = [
            f"Error Report - {self.timestamp}",
            f"Type: {self.error_type}",
            f"Message: {self.error_message}",
            f"Context: {self.context}",
            "Traceback:",
            self.traceback
        ]
        return "\n".join(lines)

async def main():
    report = ErrorReport()

    try:
        options = ClaudeAgentOptions(
            model="opus",
            max_turns=5
        )

        async for message in query(prompt="Hello", options=options):
            print(message)

    except ClaudeSDKError as e:
        report.capture(e, context={
            "model": "opus",
            "max_turns": 5,
            "prompt": "Hello"
        })

        print(report)

        # Save to file
        with open("error_report.txt", "w") as f:
            f.write(str(report))

anyio.run(main)

Error Hierarchy

Exception
└── ClaudeSDKError (base for all SDK errors)
    ├── CLIConnectionError (connection failures)
    │   └── CLINotFoundError (CLI not found/installed)
    ├── ProcessError (CLI process failures)
    ├── CLIJSONDecodeError (JSON parsing failures)
    └── MessageParseError (message parsing failures)

Best Practices

Specific Error Handling: Catch specific error types when you need different handling logic.

Catch-All Fallback: Always have a ClaudeSDKError catch-all for unexpected errors.

Error Context: Use error attributes (exit_code, stderr, line, data) for debugging.

Retry Logic: Implement retry logic for transient errors (CLIConnectionError, ProcessError).

Logging: Log errors with full context for debugging and monitoring.

Graceful Degradation: Fall back to alternative models or options on error.

Cleanup: Always clean up resources in finally blocks or use context managers.

Validation: Validate options before queries to catch configuration errors early.

Install with Tessl CLI

npx tessl i tessl/pypi-claude-agent-sdk@0.1.3

docs

reference

advanced.md

client.md

configuration.md

errors.md

hooks.md

mcp.md

messages.md

permissions.md

query.md

sandbox.md

index.md

tile.json