claude-max 0.0.10

Python SDK for Claude Code - Enhanced with Claude Max capabilities

Claude Code SDK for Python

Python SDK for Claude Code with WebSocket server and multi-agent system support. See the Claude Code SDK documentation for more information.

New Features

• OAuth Authentication: Claude Code Max users can authenticate without API keys
• WebSocket Server: Real-time bidirectional communication with streaming support
• Agent System Integration: Build multi-agent applications with specialized AI agents
• Tool Management API: Discover, create, and execute tools from a registry
• Enhanced Error Handling: Comprehensive error types and recovery mechanisms
• Type Safety: Full type hints and dataclass-based message types

Installation

# Basic installation
pip install claude-max

# With WebSocket server support
pip install claude-max[websocket]

# With agent system (separate package)
pip install claude-code-agent-system

Prerequisites:

• Python 3.10+
• Node.js
• Claude Code: npm install -g @anthropic-ai/claude-code

Quick Start

import anyio
from claude_max import query

async def main():
    async for message in query(prompt="What is 2 + 2?"):
        print(message)

anyio.run(main)

Authentication

OAuth Authentication (Claude Code Max)

Claude Code Max plan users can authenticate using OAuth without API keys:

# Login via CLI
claude-auth login

# Check authentication status
claude-auth status

Use in Python:

from claude_max import query_with_oauth

async for message in query_with_oauth(prompt="Hello Claude!"):
    print(message)

API Key Authentication

Traditional API key authentication:

export ANTHROPIC_API_KEY="your-api-key"

from claude_max import query

async for message in query(prompt="Hello Claude!"):
    print(message)

See the Authentication Guide for more details.

Usage

Basic Query

from claude_max import query, ClaudeCodeOptions, AssistantMessage, TextBlock

# Simple query
async for message in query(prompt="Hello Claude"):
    if isinstance(message, AssistantMessage):
        for block in message.content:
            if isinstance(block, TextBlock):
                print(block.text)

# With options
options = ClaudeCodeOptions(
    system_prompt="You are a helpful assistant",
    max_turns=1
)

async for message in query(prompt="Tell me a joke", options=options):
    print(message)

Using Tools

options = ClaudeCodeOptions(
    allowed_tools=["Read", "Write", "Bash"],
    permission_mode='acceptEdits'  # auto-accept file edits
)

async for message in query(
    prompt="Create a hello.py file", 
    options=options
):
    # Process tool use and results
    pass

Advanced Options

options = ClaudeCodeOptions(
    # Add additional directories for tool access
    add_dirs=["/path/to/libs", "/path/to/data"],
    
    # Skip all permission prompts (use with caution!)
    dangerously_skip_permissions=True,
    
    # Enable debug logging in CLI
    debug=True,
    
    # Control output verbosity (default: True for SDK)
    verbose=False,
    
    # Specify output format (default: "stream-json")
    output_format="json",  # "text", "json", or "stream-json"
    
    # Specify input format
    input_format="text",  # "text" or "stream-json"
)

Working Directory

from pathlib import Path

options = ClaudeCodeOptions(
    cwd="/path/to/project"  # or Path("/path/to/project")
)

API Reference

query(prompt, options=None)

Main async function for querying Claude.

Parameters:

• prompt (str): The prompt to send to Claude
• options (ClaudeCodeOptions): Optional configuration

Returns: AsyncIterator[Message] - Stream of response messages

CLI Commands

The SDK provides wrapper functions for Claude Code CLI commands:

update(cli_path=None)

Check for and install updates to Claude Code.

result = await update()
print(result)

mcp(subcommand=None, cli_path=None)

Configure and manage Model Context Protocol (MCP) servers.

# List MCP servers
servers = await mcp(["list"])

# Add MCP server
result = await mcp(["add", "my-server"])

# Remove MCP server
result = await mcp(["remove", "my-server"])

config(subcommand=None, cli_path=None)

Manage Claude Code configuration.

# Get configuration value
model = await config(["get", "model"])

# Set configuration value
result = await config(["set", "model", "claude-opus-4"])

# List all config
all_config = await config(["list"])

doctor(cli_path=None)

Check health of Claude Code auto-updater.

health = await doctor()
print(health)

version(cli_path=None)

Get Claude Code version.

ver = await version()
print(f"Claude Code version: {ver}")

Types

See src/claude_max/types.py for complete type definitions:

• ClaudeCodeOptions - Configuration options
• AssistantMessage, UserMessage, SystemMessage, ResultMessage - Message types
• TextBlock, ToolUseBlock, ToolResultBlock - Content blocks

Error Handling

from claude_max import (
    ClaudeSDKError,      # Base error
    CLINotFoundError,    # Claude Code not installed
    CLIConnectionError,  # Connection issues
    ProcessError,        # Process failed
    CLIJSONDecodeError,  # JSON parsing issues
)

try:
    async for message in query(prompt="Hello"):
        pass
except CLINotFoundError:
    print("Please install Claude Code")
except ProcessError as e:
    print(f"Process failed with exit code: {e.exit_code}")
except CLIJSONDecodeError as e:
    print(f"Failed to parse response: {e}")

See src/claude_max/_errors.py for all error types.

Troubleshooting

Common Issues and Solutions

1. CLINotFoundError: Claude Code not found

Problem: The SDK cannot find the Claude Code CLI.

Solutions:

# Install Claude Code globally
npm install -g @anthropic-ai/claude-code

# Verify installation
claude --version

If Node.js is not installed:

• macOS: brew install node or download from nodejs.org
• Linux: Use your package manager (e.g., sudo apt install nodejs npm)
• Windows: Download installer from nodejs.org

2. ProcessError: Exit code 1

Problem: The Claude Code process failed to start or crashed.

Common causes and solutions:

• Missing API key: Set your Anthropic API key:

  export ANTHROPIC_API_KEY="your-api-key"
  

• Invalid configuration: Check your ClaudeCodeOptions for typos or invalid values

• Permission issues: Ensure you have read/write permissions in the working directory

Debug steps:

try:
    async for message in query(prompt="test"):
        pass
except ProcessError as e:
    print(f"Exit code: {e.exit_code}")
    print(f"Error details: {e.stderr}")

3. CLIConnectionError: Failed to connect

Problem: Cannot establish connection to Claude Code CLI.

Solutions:

• Check if another instance is already running
• Ensure sufficient system resources (RAM, disk space)
• Try with a simple prompt first to isolate the issue

4. CLIJSONDecodeError: Invalid JSON response

Problem: Received malformed JSON from the CLI.

Common causes:

• Version mismatch between SDK and CLI
• Corrupted installation

Solutions:

# Update both SDK and CLI
pip install --upgrade claude-max
npm update -g @anthropic-ai/claude-code

5. TimeoutError during long operations

Problem: Operations taking too long and timing out.

Solution: Wrap your code with custom timeout:

import asyncio

async def long_operation():
    async with asyncio.timeout(300):  # 5 minutes
        async for message in query(prompt="Complex task..."):
            # Process messages
            pass

6. Permission denied errors

Problem: Cannot read/write files due to permissions.

Solutions:

• Run with appropriate user permissions
• Set working directory to a writable location:

  options = ClaudeCodeOptions(cwd="/path/to/writable/directory")
  

• Use permission_mode='bypassPermissions' (use with caution)

7. Rate limiting errors

Problem: Too many requests to the API.

Solutions:

• Implement exponential backoff:

  import asyncio
  
  async def query_with_retry(prompt, max_retries=3):
      for attempt in range(max_retries):
          try:
              async for message in query(prompt=prompt):
                  yield message
              break
          except ProcessError as e:
              if "rate limit" in str(e).lower() and attempt < max_retries - 1:
                  await asyncio.sleep(2 ** attempt)
              else:
                  raise
  

8. WebSocket connection issues

Problem: Cannot connect to WebSocket server.

Common issues:

• Port already in use
• Firewall blocking connections
• CORS issues in browser

Solutions:

# Use a different port
server.run(host="0.0.0.0", port=8080)

# Check if port is in use
import socket
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
result = sock.connect_ex(('localhost', 8000))
if result == 0:
    print("Port 8000 is already in use")

Debugging Tips

Enable verbose logging

import logging

# Enable debug logging
logging.basicConfig(level=logging.DEBUG)

# The SDK will now log:
# - CLI discovery process
# - Connection attempts
# - Message parsing
# - Error details

Check SDK and CLI versions

import claude_max
print(f"SDK version: {claude_max.__version__}")

claude --version

Inspect message flow

async for message in query(prompt="Debug test"):
    print(f"Message type: {type(message).__name__}")
    print(f"Message content: {message}")
    
    if hasattr(message, 'content'):
        for block in message.content:
            print(f"  Block type: {type(block).__name__}")

Test with minimal configuration

# Start with the simplest possible query
async for message in query(prompt="Hello"):
    print(message)

# If that works, gradually add complexity
options = ClaudeCodeOptions(allowed_tools=["Read"])
async for message in query(prompt="Read README.md", options=options):
    print(message)

Getting Help

If you're still experiencing issues:

1. Check the Claude Code documentation
2. Search existing GitHub issues
3. Create a new issue with:
   • Python version (python --version)
   • SDK version (pip show claude-max)
   • CLI version (claude --version)
   • Minimal code to reproduce the issue
   • Full error message and stack trace

Best Practices

1. Resource Management

Always use async context managers or ensure proper cleanup:

# Good: Automatic cleanup
async def process_files():
    async for message in query(prompt="Process files"):
        # SDK handles cleanup automatically
        pass

# For long-running applications
import asyncio

async def main():
    try:
        async for message in query(prompt="Long task"):
            process_message(message)
    except KeyboardInterrupt:
        # Graceful shutdown
        print("Shutting down...")

2. Error Handling Strategy

Implement comprehensive error handling:

from claude_max import (
    query, 
    CLINotFoundError,
    ProcessError,
    CLIConnectionError
)

async def safe_query(prompt: str, max_retries: int = 3):
    """Query with automatic retry and error handling."""
    for attempt in range(max_retries):
        try:
            async for message in query(prompt=prompt):
                yield message
            break
        except CLINotFoundError:
            # Can't recover from this
            raise
        except (CLIConnectionError, ProcessError) as e:
            if attempt < max_retries - 1:
                await asyncio.sleep(2 ** attempt)
                continue
            raise

3. Tool Usage Optimization

Be specific with tool permissions:

# Good: Only request needed tools
options = ClaudeCodeOptions(
    allowed_tools=["Read", "Write"],  # Specific tools
    disallowed_tools=["Bash"],        # Explicitly disallow
)

# Better: Use appropriate permission mode
options = ClaudeCodeOptions(
    allowed_tools=["Read", "Write", "Edit"],
    permission_mode="acceptEdits",  # Auto-accept safe operations
)

# Avoid: Too permissive
# options = ClaudeCodeOptions(permission_mode="bypassPermissions")

4. Message Processing Patterns

Process messages efficiently:

from claude_max import AssistantMessage, TextBlock, ToolUseBlock

async def process_claude_response(prompt: str):
    """Process different message types appropriately."""
    tool_uses = []
    text_responses = []
    
    async for message in query(prompt=prompt):
        if isinstance(message, AssistantMessage):
            for block in message.content:
                if isinstance(block, TextBlock):
                    text_responses.append(block.text)
                elif isinstance(block, ToolUseBlock):
                    tool_uses.append({
                        "tool": block.name,
                        "input": block.input
                    })
    
    return {
        "text": "\n".join(text_responses),
        "tools_used": tool_uses
    }

5. Session Management

For multi-turn conversations:

# Continue previous session
options = ClaudeCodeOptions(continue_conversation=True)

# Or save and resume specific sessions
session_id = None

async def chat(user_input: str):
    global session_id
    
    options = ClaudeCodeOptions(
        resume=session_id if session_id else None,
        max_turns=10  # Prevent runaway conversations
    )
    
    async for message in query(prompt=user_input, options=options):
        if isinstance(message, ResultMessage):
            session_id = message.session_id
        yield message

6. Working Directory Best Practices

Always set explicit working directories:

from pathlib import Path

# Good: Explicit, absolute path
options = ClaudeCodeOptions(
    cwd=Path("/home/user/projects/myapp").absolute()
)

# Good: Relative to script location
script_dir = Path(__file__).parent
options = ClaudeCodeOptions(
    cwd=script_dir / "workspace"
)

# Avoid: Implicit current directory
# options = ClaudeCodeOptions()  # Uses wherever script is run from

7. Performance Optimization

For better performance:

# Batch operations when possible
prompt = """
Please perform these tasks:
1. Read all Python files in the src/ directory
2. Analyze the code structure
3. Generate a summary report
"""

# Instead of multiple queries
# DON'T: Multiple round trips
# for file in files:
#     async for msg in query(f"Read {file}"):
#         ...

# DO: Single comprehensive request
async for message in query(prompt=prompt, options=options):
    # Process all results at once
    pass

8. Logging and Monitoring

Implement proper logging:

import logging
from claude_max import ResultMessage

# Configure logging
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)

logger = logging.getLogger(__name__)

async def monitored_query(prompt: str):
    """Query with monitoring."""
    start_time = time.time()
    
    try:
        async for message in query(prompt=prompt):
            if isinstance(message, ResultMessage):
                logger.info(
                    f"Query completed - "
                    f"Cost: ${message.cost_usd:.4f}, "
                    f"Duration: {message.duration_ms}ms, "
                    f"Session: {message.session_id}"
                )
            yield message
    except Exception as e:
        logger.error(f"Query failed: {e}", exc_info=True)
        raise
    finally:
        duration = time.time() - start_time
        logger.info(f"Total duration: {duration:.2f}s")

9. Security Considerations

Never expose sensitive information:

import os

# Good: Use environment variables
options = ClaudeCodeOptions(
    system_prompt="Use the API key from environment"
)

# Set via environment
os.environ["ANTHROPIC_API_KEY"] = "your-key"

# Avoid: Hardcoding secrets
# DON'T: options = ClaudeCodeOptions(
#     system_prompt="Use API key: sk-ant-..."
# )

# Good: Sanitize file paths
safe_path = Path(user_input).resolve()
if not safe_path.is_relative_to(allowed_directory):
    raise ValueError("Access denied")

10. Testing Your Integration

Write testable code:

# Make your code testable
async def analyze_code(
    file_path: str,
    query_func=query  # Injectable for testing
):
    """Analyze code with injectable query function."""
    prompt = f"Analyze the code in {file_path}"
    
    async for message in query_func(prompt=prompt):
        # Process message
        pass

# In tests
async def mock_query(prompt, options=None):
    """Mock query for testing."""
    yield AssistantMessage(content=[
        TextBlock(text="Mock analysis complete")
    ])

# Test
async def test_analyze():
    await analyze_code("test.py", query_func=mock_query)

11. Handling Large Operations

For operations that might take a long time:

import asyncio
from contextlib import asynccontextmanager

@asynccontextmanager
async def timeout_query(prompt: str, timeout_seconds: int = 300):
    """Query with timeout and cleanup."""
    task = None
    try:
        async with asyncio.timeout(timeout_seconds):
            task = asyncio.create_task(collect_messages(prompt))
            yield task
    except asyncio.TimeoutError:
        logger.warning(f"Query timed out after {timeout_seconds}s")
        raise
    finally:
        if task and not task.done():
            task.cancel()

async def collect_messages(prompt: str):
    """Collect all messages from a query."""
    messages = []
    async for message in query(prompt=prompt):
        messages.append(message)
    return messages

12. Common Pitfalls to Avoid

1. Don't use synchronous code in async context:

   # Bad
   time.sleep(1)  # Blocks event loop
   
   # Good
   await asyncio.sleep(1)
   

2. Don't ignore error messages:

   # Bad
   try:
       async for msg in query(prompt="..."):
           pass
   except Exception:
       pass  # Silent failure
   
   # Good
   except Exception as e:
       logger.error(f"Query failed: {e}")
       raise
   

3. Don't leak resources:

   # Bad
   messages = []
   async for msg in query(prompt="..."):
       messages.append(msg)
       if len(messages) > 1000000:  # Memory leak
           break
   
   # Good
   async for msg in query(prompt="..."):
       process_message(msg)  # Process and discard
   

Available Tools

See the Claude Code documentation for a complete list of available tools.

WebSocket Server

The SDK includes a WebSocket server for real-time communication:

from claude_max.websocket_server import EnhancedClaudeWebSocketServer

server = EnhancedClaudeWebSocketServer()
server.run(host="0.0.0.0", port=8000)

Connect from JavaScript:

const ws = new WebSocket('ws://localhost:8000/ws');
ws.send(JSON.stringify({
    type: 'query',
    prompt: 'Hello Claude!',
    options: { allowed_tools: ['Read', 'Write'] }
}));

See docs/websocket-server.md for full documentation.

Agent System Integration

Build multi-agent applications using the separate agent system package:

from claude_max import query, ClaudeCodeOptions
from claude_code_agent_system import BaseAgent

class CustomAgent(BaseAgent):
    async def process_with_claude(self, prompt: str):
        options = ClaudeCodeOptions(allowed_tools=["Read", "Write"])
        async for message in query(prompt=prompt, options=options):
            # Process responses
            pass

See docs/agent-system.md for integration guide.

Examples

• examples/quick_start.py - Basic SDK usage
• examples/websocket_ui_server.py - WebSocket server with UI
• examples/agent_sdk_integration.py - Agent system integration
• agent_system/ - Complete agent system implementation

Documentation

• WebSocket Server Guide
• Agent System Integration
• API Reference

Contributing

Contributions are welcome! Please read our contributing guidelines and submit pull requests to our repository.

License

MIT