Error Handling

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

from claude_agent_sdk import query, ClaudeSDKError

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."""

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"Failed to connect to Claude Code: {e}")

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
    ):
        """
        Initialize CLINotFoundError.

        Args:
            message: Error message
            cli_path: Path where CLI was expected
        """

from claude_agent_sdk import query, ClaudeAgentOptions, CLINotFoundError

try:
    options = ClaudeAgentOptions(cli_path="/custom/path/claude")
    async for msg in query(prompt="Hello", options=options):
        print(msg)
except CLINotFoundError as e:
    print(f"Claude Code CLI not found: {e}")
    # The cli_path is included in the error message

from claude_agent_sdk import ClaudeAgentOptions, CLINotFoundError
import shutil

try:
    # Try custom path first
    options = ClaudeAgentOptions(cli_path="/usr/local/bin/claude")
    async for msg in query(prompt="Hello", options=options):
        print(msg)
except CLINotFoundError:
    # Fall back to default bundled CLI
    print("Custom CLI not found, using bundled version")
    async for msg in query(prompt="Hello"):
        print(msg)

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

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

        Args:
            message: Error message
            exit_code: Process exit code
            stderr: Process stderr output
        """

    exit_code: int | None
    stderr: str | None

from claude_agent_sdk import query, ClaudeAgentOptions, ProcessError

try:
    options = ClaudeAgentOptions(
        allowed_tools=["Bash"],
        permission_mode="bypassPermissions"
    )
    async for msg in query(prompt="Run invalid command", options=options):
        print(msg)
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}")

from claude_agent_sdk import query, ProcessError

try:
    async for msg in query(prompt="Hello"):
        print(msg)
except ProcessError as e:
    if e.exit_code == 126:
        print("Permission denied - check file/directory permissions")
    elif e.exit_code == 130:
        print("Operation interrupted by user")
    else:
        print(f"Process failed with exit code {e.exit_code}")

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

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

        Args:
            line: Line that failed to parse
            original_error: Original JSON decode error
        """

    line: str
    original_error: Exception

from claude_agent_sdk import query, CLIJSONDecodeError

try:
    async for msg in query(prompt="Hello"):
        print(msg)
except CLIJSONDecodeError as e:
    print(f"Failed to parse CLI output: {e}")
    print(f"Problematic line: {e.line[:100]}")  # First 100 chars
    print(f"Original error: {e.original_error}")

from claude_agent_sdk import query, ClaudeAgentOptions, CLIJSONDecodeError

def debug_stderr(line: str):
    print(f"[DEBUG] CLI stderr: {line}")

try:
    options = ClaudeAgentOptions(stderr=debug_stderr)
    async for msg in query(prompt="Hello", options=options):
        print(msg)
except CLIJSONDecodeError as e:
    print(f"JSON parse error: {e}")
    # Check if stderr callback revealed issues

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

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

        Args:
            message: Error message
            data: Raw message data if available
        """

    data: dict[str, Any] | None

from claude_agent_sdk import query, MessageParseError

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

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

async def safe_query(prompt: str, options: ClaudeAgentOptions | None = None):
    """Query with comprehensive error handling."""
    try:
        async for msg in query(prompt=prompt, options=options):
            yield msg

    except CLINotFoundError as e:
        print(f"ERROR: Claude Code CLI not found: {e}")
        print("  Please check installation or cli_path option")

    except ProcessError as e:
        print(f"ERROR: CLI process failed: {e}")
        if e.exit_code:
            print(f"  Exit code: {e.exit_code}")
        if e.stderr:
            print(f"  Error output: {e.stderr[:500]}")  # First 500 chars

    except CLIJSONDecodeError as e:
        print(f"ERROR: Failed to parse CLI output: {e}")
        print(f"  Problematic line: {e.line[:200]}")
        print(f"  This may indicate a version mismatch")

    except MessageParseError as e:
        print(f"ERROR: Failed to parse message: {e}")
        if e.data:
            print(f"  Message type: {e.data.get('type', 'unknown')}")

    except CLIConnectionError as e:
        print(f"ERROR: Failed to connect to Claude Code: {e}")
        print("  Check that Claude Code is properly installed")

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

    except Exception as e:
        print(f"ERROR: Unexpected error: {type(e).__name__}: {e}")

# Usage
async def main():
    async for msg in safe_query("Hello, Claude!"):
        print(msg)

anyio.run(main)

from claude_agent_sdk import query, ProcessError, CLIConnectionError
import anyio

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

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

# Usage
async def main():
    async for msg in query_with_retry("Hello", max_retries=3):
        print(msg)

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

async def query_with_fallback(prompt: str):
    """Query with fallback to simpler configuration."""
    # Try with full configuration
    try:
        options = ClaudeAgentOptions(
            allowed_tools=["Read", "Write", "Bash"],
            permission_mode="acceptEdits",
            model="claude-opus-4-1-20250805"
        )
        async for msg in query(prompt=prompt, options=options):
            yield msg
        return

    except ProcessError as e:
        print(f"Full config failed: {e}, trying with reduced tools...")

    # Fallback to fewer tools
    try:
        options = ClaudeAgentOptions(
            allowed_tools=["Read"],
            model="claude-sonnet-4-5-20250929"
        )
        async for msg in query(prompt=prompt, options=options):
            yield msg
        return

    except ProcessError as e:
        print(f"Reduced tools failed: {e}, trying basic query...")

    # Fallback to basic query
    async for msg in query(prompt=prompt):
        yield msg

from claude_agent_sdk import query, ClaudeAgentOptions, ClaudeSDKError
import traceback
import sys

async def query_with_context(prompt: str, options: ClaudeAgentOptions | None = None):
    """Query with detailed error context collection."""
    try:
        async for msg in query(prompt=prompt, options=options):
            yield msg

    except ClaudeSDKError as e:
        # Collect error context
        error_info = {
            "error_type": type(e).__name__,
            "error_message": str(e),
            "traceback": traceback.format_exc(),
            "prompt": prompt[:100],  # First 100 chars
            "python_version": sys.version,
        }

        # Add error-specific context
        if hasattr(e, "exit_code"):
            error_info["exit_code"] = e.exit_code
        if hasattr(e, "stderr"):
            error_info["stderr"] = e.stderr[:500] if e.stderr else None
        if hasattr(e, "cli_path"):
            error_info["cli_path"] = e.cli_path
        if hasattr(e, "data"):
            error_info["raw_data"] = e.data

        # Log or report error context
        print("Error context:")
        for key, value in error_info.items():
            print(f"  {key}: {value}")

        raise

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

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

    except CLINotFoundError:
        print("""
╭─────────────────────────────────────────────╮
│ Claude Code Not Found                       │
├─────────────────────────────────────────────┤
│ The Claude Code CLI could not be found.    │
│                                             │
│ Please ensure:                              │
│  • Claude Code is properly installed        │
│  • The bundled CLI is not corrupted         │
│  • You have the correct cli_path configured │
│                                             │
│ Visit: https://docs.anthropic.com/claude   │
╰─────────────────────────────────────────────╯
        """)

    except ProcessError as e:
        if e.exit_code == 126:
            print("""
╭─────────────────────────────────────────────╮
│ Permission Denied                           │
├─────────────────────────────────────────────┤
│ The CLI process encountered a permission    │
│ error.                                      │
│                                             │
│ This may be caused by:                      │
│  • File/directory access restrictions       │
│  • Tool permission mode settings            │
│  • Operating system security policies       │
│                                             │
│ Try adjusting permission_mode or working    │
│ directory settings.                         │
╰─────────────────────────────────────────────╯
            """)
        else:
            print(f"""
╭─────────────────────────────────────────────╮
│ Process Error                               │
├─────────────────────────────────────────────┤
│ The Claude Code CLI process failed.        │
│                                             │
│ Exit code: {e.exit_code or 'Unknown'}
│                                             │
│ Error output:                               │
│ {(e.stderr or 'No error output')[:200]}
╰─────────────────────────────────────────────╯
            """)

    except CLIConnectionError:
        print("""
╭─────────────────────────────────────────────╮
│ Connection Error                            │
├─────────────────────────────────────────────┤
│ Failed to establish connection to Claude    │
│ Code CLI.                                   │
│                                             │
│ This may indicate:                          │
│  • CLI initialization failure               │
│  • Invalid configuration options            │
│  • Resource constraints                     │
│                                             │
│ Check logs for more details.                │
╰─────────────────────────────────────────────╯
        """)

from claude_agent_sdk import ClaudeSDKClient, ClaudeSDKError
import anyio

async def safe_client_session(prompt: str):
    """Client session with proper error handling."""
    client = None
    try:
        client = ClaudeSDKClient()
        await client.connect()

        await client.query(prompt)
        async for msg in client.receive_response():
            print(msg)

    except ClaudeSDKError as e:
        print(f"Error during session: {e}")
        # Handle error appropriately

    finally:
        # Always clean up
        if client:
            try:
                await client.disconnect()
            except Exception as e:
                print(f"Error during cleanup: {e}")

# Better: use context manager
async def safe_client_session_v2(prompt: str):
    """Client session with context manager."""
    try:
        async with ClaudeSDKClient() as client:
            await client.query(prompt)
            async for msg in client.receive_response():
                print(msg)
    except ClaudeSDKError as e:
        print(f"Error during session: {e}")

from claude_agent_sdk import (
    query, ClaudeAgentOptions, ClaudeSDKError
)
import pytest

@pytest.mark.asyncio
async def test_invalid_cli_path():
    """Test handling of invalid CLI path."""
    with pytest.raises(CLINotFoundError):
        options = ClaudeAgentOptions(cli_path="/nonexistent/path")
        async for msg in query(prompt="Hello", options=options):
            pass

@pytest.mark.asyncio
async def test_error_recovery():
    """Test error recovery and retry."""
    attempts = 0
    max_attempts = 3

    while attempts < max_attempts:
        try:
            async for msg in query(prompt="Hello"):
                break  # Success
        except ClaudeSDKError:
            attempts += 1
            if attempts >= max_attempts:
                raise

    assert attempts < max_attempts, "Should succeed within retry limit"

import logging
from claude_agent_sdk import ClaudeAgentOptions

logging.basicConfig(level=logging.DEBUG)

def stderr_handler(line: str):
    logging.debug(f"CLI stderr: {line}")

options = ClaudeAgentOptions(stderr=stderr_handler)

from claude_agent_sdk import query, ClaudeSDKError

try:
    async for msg in query(prompt="Hello"):
        print(msg)
except ClaudeSDKError as e:
    # Inspect all error attributes
    print(f"Error type: {type(e)}")
    print(f"Error message: {e}")
    print(f"Attributes: {dir(e)}")

    # Error-specific attributes
    if hasattr(e, "__dict__"):
        print(f"Instance dict: {e.__dict__}")

from claude_agent_sdk import __version__

print(f"SDK version: {__version__}")

from claude_agent_sdk import ClaudeAgentOptions

options = ClaudeAgentOptions(
    allowed_tools=["Read", "Write"],
    permission_mode="acceptEdits"
)

# Validate before use
assert options.permission_mode in ["default", "acceptEdits", "plan", "bypassPermissions"]
assert all(isinstance(tool, str) for tool in options.allowed_tools)