Error Types

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

    All SDK-specific exceptions inherit from this class, allowing you
    to catch all SDK errors with a single except clause.

    Example:
        try:
            async for msg in query(prompt="Hello"):
                print(msg)
        except ClaudeSDKError as e:
            print(f"SDK error: {e}")
    """

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

    This error occurs when the SDK cannot establish a connection to the
    Claude Code CLI process. This can happen if:
    - The CLI is not installed
    - The CLI path is incorrect
    - Network issues (for remote transports)
    - Permission issues

    Inherits from: ClaudeSDKError
    """

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

    This specific connection error occurs when the SDK cannot locate the
    Claude Code CLI executable. This typically means Claude Code is not
    installed or not in the system PATH.

    Inherits from: CLIConnectionError

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

    Attributes:
        cli_path: The path where the CLI was expected (if known)
    """

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

        Args:
            message: Error message. Defaults to "Claude Code not found"
            cli_path: Optional path where CLI was expected. If provided,
                will be appended to the error message.

        Example:
            CLINotFoundError("Custom message", "/usr/local/bin/claude")
        """

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

    This error occurs when the Claude Code CLI process exits with an error
    or fails during execution. Contains details about the failure including
    exit code and stderr output.

    Inherits from: ClaudeSDKError

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

    Attributes:
        exit_code: Process exit code (if available)
        stderr: Standard error output from the process (if available)
    """

    def __init__(
        self,
        message: str,
        exit_code: int | None = None,
        stderr: str | None = None
    ):
        """
        Initialize ProcessError.

        Args:
            message: Error description
            exit_code: Optional process exit code
            stderr: Optional stderr output from process

        Example:
            ProcessError(
                "Process failed",
                exit_code=1,
                stderr="Error: Invalid argument"
            )

        Notes:
            - Exit code and stderr are automatically appended to the message
            - The formatted message includes exit code if available
            - Stderr output is included in formatted message if available
        """

    exit_code: int | None
    """Process exit code.

    The exit code returned by the CLI process. None if process didn't
    exit normally or exit code is unknown.

    Common values:
    - 0: Success (but wouldn't raise ProcessError)
    - 1: General error
    - 2: Misuse of shell command
    - 126: Command cannot execute
    - 127: Command not found
    - 130: Terminated by Ctrl+C
    """

    stderr: str | None
    """Standard error output.

    The stderr output from the CLI process. None if no stderr was captured
    or process didn't produce stderr output.

    This typically contains error messages, warnings, and diagnostic
    information from the CLI.
    """

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

    This error occurs when the CLI produces output that cannot be parsed
    as JSON. This usually indicates a protocol error or corrupted output.

    Inherits from: ClaudeSDKError

    Constructor:
        __init__(line: str, original_error: Exception)

    Attributes:
        line: The line that failed to parse
        original_error: The underlying JSON decode exception
    """

    def __init__(self, line: str, original_error: Exception):
        """
        Initialize CLIJSONDecodeError.

        Args:
            line: The line that failed to parse (truncated to 100 chars in message)
            original_error: The original exception from JSON parsing

        Example:
            CLIJSONDecodeError(
                "{invalid json",
                json.JSONDecodeError("msg", "doc", 0)
            )

        Notes:
            - Error message shows first 100 characters of problematic line
            - Original error is preserved for debugging
        """

    line: str
    """Line that failed to parse.

    The complete line from CLI output that could not be decoded as JSON.
    This is the raw string that caused the parsing error.
    """

    original_error: Exception
    """Original exception.

    The underlying exception that occurred during JSON parsing. Usually
    a json.JSONDecodeError with details about what went wrong.

    Useful for debugging the specific parsing issue.
    """

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

    This error occurs when the CLI output is valid JSON but doesn't match
    the expected message structure. This indicates a protocol mismatch or
    unexpected message format.

    Inherits from: ClaudeSDKError

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

    Attributes:
        data: The message data that failed to parse (if available)
    """

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

        Args:
            message: Error description
            data: Optional message data that failed to parse

        Example:
            MessageParseError(
                "Unknown message type",
                {"type": "unknown", "data": {...}}
            )

        Notes:
            - Message describes what went wrong during parsing
            - Data contains the raw message dict for debugging
        """

    data: dict[str, Any] | None
    """Message data.

    The parsed JSON data that couldn't be converted to a proper message
    object. None if data is not available.

    This is the dictionary that was successfully parsed as JSON but
    doesn't match expected message schemas.
    """

from claude_agent_sdk import query, ClaudeSDKError

try:
    async for msg in query(prompt="Hello Claude"):
        print(msg)
except ClaudeSDKError as e:
    print(f"SDK error occurred: {e}")

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

try:
    async for msg in query(prompt="Hello"):
        print(msg)

except CLINotFoundError as e:
    print("Claude Code is not installed or not in PATH")
    print("Please install from: https://claude.com/code")

except CLIConnectionError as e:
    print(f"Cannot connect to Claude Code: {e}")
    print("Make sure Claude Code is running")

except ProcessError as e:
    print(f"CLI process failed: {e}")
    if e.exit_code:
        print(f"Exit code: {e.exit_code}")
    if e.stderr:
        print(f"Error output: {e.stderr}")

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

from claude_agent_sdk import query, CLINotFoundError

async def check_claude_installed():
    """Check if Claude Code is installed."""
    try:
        async for msg in query(prompt="test"):
            # If we get here, Claude is installed
            return True
    except CLINotFoundError:
        return False

if await check_claude_installed():
    print("Claude Code is installed")
else:
    print("Please install Claude Code")

from claude_agent_sdk import query, ProcessError

try:
    async for msg in query(prompt="Hello"):
        print(msg)

except ProcessError as e:
    print("Process Error Details:")
    print(f"  Message: {e}")
    print(f"  Exit Code: {e.exit_code}")
    print(f"  Stderr: {e.stderr}")

    # Check specific exit codes
    if e.exit_code == 127:
        print("  Diagnosis: Command not found")
    elif e.exit_code == 126:
        print("  Diagnosis: Command cannot execute (permission issue?)")
    elif e.exit_code == 130:
        print("  Diagnosis: Process interrupted by user")

import anyio
from claude_agent_sdk import query, CLIConnectionError, ClaudeSDKError

async def query_with_retry(prompt: str, max_retries: int = 3):
    """Query with automatic retry on connection errors."""
    for attempt in range(max_retries):
        try:
            async for msg in query(prompt=prompt):
                print(msg)
            return  # Success

        except CLIConnectionError as e:
            if attempt < max_retries - 1:
                print(f"Connection failed (attempt {attempt + 1}/{max_retries})")
                print("Retrying in 2 seconds...")
                await anyio.sleep(2)
            else:
                print("Max retries reached")
                raise

        except ClaudeSDKError:
            # Don't retry other errors
            raise

await query_with_retry("Hello Claude")

from claude_agent_sdk import query, CLIJSONDecodeError

try:
    async for msg in query(prompt="Hello"):
        print(msg)

except CLIJSONDecodeError as e:
    print("Failed to parse CLI output")
    print(f"Problematic line: {e.line[:200]}")  # Show first 200 chars
    print(f"Original error: {e.original_error}")
    print("This might indicate a CLI version mismatch")

from claude_agent_sdk import query, MessageParseError

try:
    async for msg in query(prompt="Hello"):
        print(msg)

except MessageParseError as e:
    print("Failed to parse message structure")
    print(f"Error: {e}")
    if e.data:
        print(f"Raw data: {e.data}")
    print("This might indicate a protocol version mismatch")

import logging
from claude_agent_sdk import query, ClaudeSDKError, ProcessError

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

try:
    async for msg in query(prompt="Hello"):
        print(msg)

except ProcessError as e:
    logger.error(
        "Process error",
        extra={
            "exit_code": e.exit_code,
            "stderr": e.stderr,
            "message": str(e)
        }
    )

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

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

async def robust_query(prompt: str):
    """Query with error recovery strategies."""
    try:
        async for msg in query(prompt=prompt):
            print(msg)

    except CLINotFoundError:
        print("Installing Claude Code would fix this...")
        # Could trigger installation process here
        raise

    except ProcessError as e:
        if e.exit_code == 1 and e.stderr and "permission" in e.stderr.lower():
            print("Try running with different permissions")
        else:
            print(f"Process failed: {e}")
        raise

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

await robust_query("Hello Claude")

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

async def query_with_friendly_errors(prompt: str):
    """Query with user-friendly error messages."""
    try:
        async for msg in query(prompt=prompt):
            print(msg)

    except CLINotFoundError:
        print("\nClaude Code Not Found")
        print("=" * 50)
        print("Claude Code needs to be installed to use this SDK.")
        print("\nInstallation instructions:")
        print("1. Visit https://claude.com/code")
        print("2. Download and install Claude Code")
        print("3. Make sure 'claude' is in your PATH")

    except CLIConnectionError as e:
        print("\nConnection Error")
        print("=" * 50)
        print("Could not connect to Claude Code.")
        print(f"\nDetails: {e}")
        print("\nPossible solutions:")
        print("- Check if Claude Code is running")
        print("- Try restarting Claude Code")
        print("- Check your network connection")

    except ProcessError as e:
        print("\nProcess Error")
        print("=" * 50)
        print("Claude Code process encountered an error.")
        if e.exit_code:
            print(f"Exit code: {e.exit_code}")
        if e.stderr:
            print(f"Error details: {e.stderr}")

    except ClaudeSDKError as e:
        print("\nUnexpected Error")
        print("=" * 50)
        print(f"An unexpected error occurred: {e}")
        print("Please report this issue if it persists")

await query_with_friendly_errors("Hello")

from claude_agent_sdk import query, ClaudeSDKError

class ApplicationError(Exception):
    """Custom application error."""
    pass

try:
    async for msg in query(prompt="Hello"):
        print(msg)

except ClaudeSDKError as e:
    # Chain SDK error with application error
    raise ApplicationError("Failed to query Claude") from e

from collections import defaultdict
from claude_agent_sdk import (
    query, CLINotFoundError, CLIConnectionError,
    ProcessError, ClaudeSDKError
)

error_counts = defaultdict(int)

async def query_with_metrics(prompt: str):
    """Track error types for monitoring."""
    try:
        async for msg in query(prompt=prompt):
            print(msg)

    except CLINotFoundError as e:
        error_counts["cli_not_found"] += 1
        raise

    except CLIConnectionError as e:
        error_counts["connection_error"] += 1
        raise

    except ProcessError as e:
        error_counts["process_error"] += 1
        raise

    except ClaudeSDKError as e:
        error_counts["other_sdk_error"] += 1
        raise

# Later, check metrics
print("Error statistics:")
for error_type, count in error_counts.items():
    print(f"  {error_type}: {count}")

from claude_agent_sdk import query, ProcessError, ClaudeSDKError

async def handle_errors_conditionally(prompt: str, strict: bool = False):
    """Handle errors differently based on mode."""
    try:
        async for msg in query(prompt=prompt):
            print(msg)

    except ProcessError as e:
        if strict:
            # In strict mode, fail immediately
            raise
        else:
            # In lenient mode, log and continue
            print(f"Warning: Process error occurred: {e}")
            print("Continuing anyway...")

    except ClaudeSDKError as e:
        if strict:
            raise
        else:
            print(f"Warning: SDK error: {e}")

# Strict mode
await handle_errors_conditionally("Hello", strict=True)

# Lenient mode
await handle_errors_conditionally("Hello", strict=False)

from contextlib import asynccontextmanager
from claude_agent_sdk import query, ClaudeSDKError

@asynccontextmanager
async def handle_sdk_errors():
    """Context manager for SDK error handling."""
    try:
        yield
    except ClaudeSDKError as e:
        print(f"SDK Error: {e}")
        # Log to file, send to monitoring, etc.
        raise

async def main():
    async with handle_sdk_errors():
        async for msg in query(prompt="Hello"):
            print(msg)

await main()