Python SDK for Claude Code providing simple query functions and advanced bidirectional interactive conversations with custom tool support
Comprehensive error handling with specific exception types for different failure modes. The Claude Code SDK provides detailed exception hierarchy for connection errors, process failures, JSON parsing issues, and message parsing errors.
Base exception for all Claude SDK errors.
class ClaudeSDKError(Exception):
"""Base exception for all Claude SDK errors."""Errors related to connecting to Claude Code CLI.
class CLIConnectionError(ClaudeSDKError):
"""Raised when unable to connect to Claude Code."""
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 Claude Code was expected (if known)
"""Errors related to CLI process execution and failures.
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 description
exit_code: Process exit code (if available)
stderr: Standard error output (if available)
"""
exit_code: int | None
stderr: str | NoneErrors related to parsing Claude Code CLI output and responses.
class CLIJSONDecodeError(ClaudeSDKError):
"""Raised when unable to decode JSON from CLI output."""
def __init__(self, line: str, original_error: Exception):
"""
Initialize CLIJSONDecodeError.
Args:
line: The line that failed to parse
original_error: The original JSON decode exception
"""
line: str
original_error: Exceptionfrom claude_code_sdk import (
query, ClaudeSDKError, CLINotFoundError,
CLIConnectionError, ProcessError, CLIJSONDecodeError
)
async def main():
try:
async for message in query(prompt="Hello Claude"):
print(message)
except CLINotFoundError:
print("Error: Claude Code is not installed.")
print("Please install with: npm install -g @anthropic-ai/claude-code")
except CLIConnectionError as e:
print(f"Error: Unable to connect to Claude Code: {e}")
except ProcessError as e:
print(f"Error: Claude Code process failed: {e}")
if e.exit_code:
print(f"Exit code: {e.exit_code}")
if e.stderr:
print(f"Error output: {e.stderr}")
except CLIJSONDecodeError as e:
print(f"Error: Failed to parse Claude Code response")
print(f"Problematic line: {e.line[:100]}")
print(f"Original error: {e.original_error}")
except ClaudeSDKError as e:
print(f"Error: General Claude SDK error: {e}")
except Exception as e:
print(f"Unexpected error: {e}")import asyncio
import time
from claude_code_sdk import (
query, ClaudeCodeOptions, CLINotFoundError,
CLIConnectionError, ProcessError
)
async def robust_query(prompt: str, max_retries: int = 3) -> list:
"""Query with retry logic and error handling."""
messages = []
last_error = None
for attempt in range(max_retries):
try:
async for message in query(prompt=prompt):
messages.append(message)
return messages
except CLINotFoundError as e:
print("Claude Code not found. Please install it first.")
raise e # Don't retry installation errors
except CLIConnectionError as e:
last_error = e
print(f"Connection error (attempt {attempt + 1}/{max_retries}): {e}")
if attempt < max_retries - 1:
await asyncio.sleep(2 ** attempt) # Exponential backoff
continue
except ProcessError as e:
last_error = e
# Retry on certain exit codes
if e.exit_code in [1, 2]: # Retryable errors
print(f"Process error (attempt {attempt + 1}/{max_retries}): {e}")
if attempt < max_retries - 1:
await asyncio.sleep(1)
continue
else:
# Non-retryable process error
raise e
# If we get here, all retries failed
raise last_error or Exception("All retry attempts failed")
async def main():
try:
messages = await robust_query("What is the capital of France?")
for message in messages:
print(message)
except Exception as e:
print(f"Failed after all retries: {e}")from claude_code_sdk import (
ClaudeSDKClient, ClaudeCodeOptions,
CLIConnectionError, ProcessError
)
async def main():
client = ClaudeSDKClient()
try:
await client.connect("Hello Claude")
async for msg in client.receive_response():
print(msg)
except CLIConnectionError as e:
print(f"Failed to connect: {e}")
print("Ensure Claude Code is installed and accessible")
except ProcessError as e:
print(f"Process error during conversation: {e}")
if e.exit_code == 130: # SIGINT
print("Conversation was interrupted")
elif e.stderr and "permission" in e.stderr.lower():
print("Permission denied - check tool permissions")
finally:
try:
await client.disconnect()
except Exception as disconnect_error:
print(f"Error during disconnect: {disconnect_error}")from claude_code_sdk import (
tool, create_sdk_mcp_server, ClaudeSDKClient,
ClaudeCodeOptions, ProcessError
)
@tool("divide", "Divide two numbers", {"a": float, "b": float})
async def divide_numbers(args):
"""Tool with proper error handling."""
try:
if args["b"] == 0:
return {
"content": [
{"type": "text", "text": "Error: Division by zero is not allowed"}
],
"is_error": True
}
result = args["a"] / args["b"]
return {
"content": [
{"type": "text", "text": f"Result: {result}"}
]
}
except Exception as e:
return {
"content": [
{"type": "text", "text": f"Unexpected error: {str(e)}"}
],
"is_error": True
}
async def main():
server = create_sdk_mcp_server("calculator", tools=[divide_numbers])
options = ClaudeCodeOptions(
mcp_servers={"calc": server},
allowed_tools=["mcp__calc__divide"]
)
try:
async with ClaudeSDKClient(options=options) as client:
await client.query("Divide 10 by 0")
async for msg in client.receive_response():
# Check for tool errors in messages
if hasattr(msg, 'content'):
for block in getattr(msg, 'content', []):
if hasattr(block, 'is_error') and block.is_error:
print(f"Tool reported error: {block.content}")
elif hasattr(block, 'text'):
print(f"Response: {block.text}")
except ProcessError as e:
print(f"Process error with custom tools: {e}")import logging
import sys
from claude_code_sdk import (
query, ClaudeCodeOptions, CLIJSONDecodeError,
MessageParseError
)
# Configure logging
logging.basicConfig(
level=logging.DEBUG,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
handlers=[
logging.FileHandler('claude_sdk.log'),
logging.StreamHandler(sys.stdout)
]
)
logger = logging.getLogger('claude_sdk_app')
async def main():
# Enable debug output
debug_file = open("claude_debug.log", "w")
options = ClaudeCodeOptions(
debug_stderr=debug_file,
allowed_tools=["Read", "Write", "Bash"]
)
try:
async for message in query(
prompt="Create a complex project structure",
options=options
):
logger.info(f"Received message: {type(message).__name__}")
print(message)
except CLIJSONDecodeError as e:
logger.error(f"JSON decode error: {e.line[:100]}")
logger.error(f"Original error: {e.original_error}")
# Log the problematic line for debugging
with open("failed_json.log", "w") as f:
f.write(f"Failed line: {e.line}\n")
f.write(f"Error: {e.original_error}\n")
except MessageParseError as e:
logger.error(f"Message parse error: {e}")
if e.data:
logger.error(f"Problematic data: {e.data}")
except Exception as e:
logger.exception(f"Unexpected error: {e}")
finally:
debug_file.close()from claude_code_sdk import ClaudeSDKClient, ClaudeCodeOptions
from contextlib import asynccontextmanager
@asynccontextmanager
async def claude_client_with_recovery(options=None):
"""Context manager with automatic error recovery."""
client = ClaudeSDKClient(options or ClaudeCodeOptions())
try:
await client.connect()
yield client
except CLIConnectionError as e:
print(f"Connection error: {e}")
# Attempt recovery
try:
await client.connect()
yield client
except Exception as recovery_error:
print(f"Recovery failed: {recovery_error}")
raise e
except Exception as e:
print(f"Unexpected error in context manager: {e}")
raise
finally:
try:
await client.disconnect()
except Exception as disconnect_error:
print(f"Error during cleanup: {disconnect_error}")
async def main():
try:
async with claude_client_with_recovery() as client:
await client.query("Test with error recovery")
async for msg in client.receive_response():
print(msg)
except Exception as e:
print(f"Final error: {e}")from claude_code_sdk import ClaudeCodeOptions, ClaudeSDKClient
async def validate_options(options: ClaudeCodeOptions) -> ClaudeCodeOptions:
"""Validate and fix common configuration errors."""
# Fix common tool name issues
if options.allowed_tools:
fixed_tools = []
for tool in options.allowed_tools:
if tool in ["bash", "Bash"]:
fixed_tools.append("Bash")
elif tool in ["read", "Read"]:
fixed_tools.append("Read")
elif tool in ["write", "Write"]:
fixed_tools.append("Write")
else:
fixed_tools.append(tool)
options.allowed_tools = fixed_tools
# Check for conflicting permission settings
if options.can_use_tool and options.permission_prompt_tool_name:
print("Warning: can_use_tool and permission_prompt_tool_name are mutually exclusive")
print("Setting permission_prompt_tool_name to None")
options.permission_prompt_tool_name = None
return options
async def main():
try:
# Potentially problematic configuration
options = ClaudeCodeOptions(
allowed_tools=["bash", "read", "write"], # Wrong case
can_use_tool=lambda *args: None, # Invalid callback
permission_prompt_tool_name="custom" # Conflicting setting
)
# Validate and fix
options = await validate_options(options)
async with ClaudeSDKClient(options=options) as client:
await client.query("Test with validated options")
async for msg in client.receive_response():
print(msg)
except ValueError as e:
print(f"Configuration error: {e}")
except Exception as e:
print(f"Unexpected error: {e}")CLINotFoundErrorCLIConnectionErrorProcessErrorCLIJSONDecodeError, MessageParseErrorExceptiondebug_stderr optionFor related configuration options, see Configuration and Options. For transport-specific errors, see Transport System.
Install with Tessl CLI
npx tessl i tessl/pypi-claude-code-sdk