tessl/pypi-anthropic

The official Python library for the anthropic API

—

Pending

Overview

Eval results

Files

Error Handling

Name: tessl/pypi-anthropic
Author: tessl

The Anthropic SDK provides a comprehensive exception hierarchy for handling API errors, network issues, authentication problems, and service-specific errors. This enables robust error handling and appropriate responses to different failure scenarios.

Capabilities

Exception Hierarchy

All Anthropic SDK exceptions inherit from a base exception class, providing a structured approach to error handling.

class AnthropicError(Exception):
    """Base exception for all Anthropic SDK errors"""
    pass

class APIError(AnthropicError):
    """Base class for API-related errors"""
    pass

class APIStatusError(APIError):
    """HTTP status code errors"""
    def __init__(self, message: str, *, response: httpx.Response, body: Optional[object]): ...
    
    @property
    def status_code(self) -> int: ...
    @property
    def response(self) -> httpx.Response: ...
    @property  
    def body(self) -> Optional[object]: ...

class APITimeoutError(APIError):
    """Request timeout errors"""
    pass

class APIConnectionError(APIError):
    """Network connection errors"""
    pass

class APIResponseValidationError(APIError):
    """Response validation errors"""
    pass

HTTP Status Errors

Specific exception classes for different HTTP status codes and API error types.

class BadRequestError(APIStatusError):
    """HTTP 400 Bad Request errors"""
    pass

class AuthenticationError(APIStatusError):
    """HTTP 401 Authentication errors"""
    pass

class PermissionDeniedError(APIStatusError):
    """HTTP 403 Permission Denied errors"""
    pass

class NotFoundError(APIStatusError):
    """HTTP 404 Not Found errors"""
    pass

class ConflictError(APIStatusError):
    """HTTP 409 Conflict errors"""
    pass

class UnprocessableEntityError(APIStatusError):
    """HTTP 422 Unprocessable Entity errors"""
    pass

class RateLimitError(APIStatusError):
    """HTTP 429 Rate Limit errors"""
    pass

class InternalServerError(APIStatusError):
    """HTTP 500 Internal Server errors"""
    pass

Error Response Types

Structured error response objects that provide detailed error information.

class ErrorObject(TypedDict):
    type: str
    message: str

class ErrorResponse(TypedDict):
    type: Literal["error"]
    error: ErrorObject

class APIErrorObject(TypedDict):
    type: str
    message: str

class AuthenticationError(TypedDict):
    type: Literal["authentication_error"]
    message: str

class PermissionError(TypedDict):
    type: Literal["permission_error"]
    message: str

class NotFoundError(TypedDict):
    type: Literal["not_found_error"]
    message: str

class RateLimitError(TypedDict):
    type: Literal["rate_limit_error"]
    message: str

class OverloadedError(TypedDict):
    type: Literal["overloaded_error"]
    message: str

class BillingError(TypedDict):
    type: Literal["billing_error"]
    message: str

class GatewayTimeoutError(TypedDict):
    type: Literal["gateway_timeout_error"]
    message: str

class InvalidRequestError(TypedDict):
    type: Literal["invalid_request_error"]
    message: str

Usage Examples

Basic Error Handling

from anthropic import Anthropic, APIError, RateLimitError, AuthenticationError

client = Anthropic()

try:
    message = client.messages.create(
        model="claude-sonnet-4-20250514",
        max_tokens=1024,
        messages=[
            {"role": "user", "content": "Hello!"}
        ]
    )
    print(message.content[0].text)
    
except AuthenticationError as e:
    print(f"Authentication failed: {e}")
    print("Please check your API key")
    
except RateLimitError as e:
    print(f"Rate limit exceeded: {e}")
    print("Please wait before making more requests")
    
except APIError as e:
    print(f"API error occurred: {e}")
    print(f"Status code: {e.status_code}")
    
except Exception as e:
    print(f"Unexpected error: {e}")

Comprehensive Error Handler

import time
import logging
from typing import Optional, Any

def handle_anthropic_request(
    request_func: callable,
    *args,
    max_retries: int = 3,
    backoff_factor: float = 1.0,
    **kwargs
) -> Optional[Any]:
    """
    Execute an Anthropic API request with comprehensive error handling and retries
    """
    
    for attempt in range(max_retries):
        try:
            return request_func(*args, **kwargs)
            
        except AuthenticationError as e:
            logging.error(f"Authentication error: {e}")
            print("❌ Authentication failed. Please check your API key.")
            return None
            
        except PermissionDeniedError as e:
            logging.error(f"Permission denied: {e}")
            print("❌ Permission denied. Check your account permissions.")
            return None
            
        except RateLimitError as e:
            logging.warning(f"Rate limit exceeded (attempt {attempt + 1}): {e}")
            if attempt < max_retries - 1:
                wait_time = backoff_factor * (2 ** attempt)
                print(f"⏳ Rate limited. Waiting {wait_time} seconds...")
                time.sleep(wait_time)
                continue
            else:
                print("❌ Rate limit exceeded. Max retries reached.")
                return None
                
        except OverloadedError as e:
            logging.warning(f"Service overloaded (attempt {attempt + 1}): {e}")
            if attempt < max_retries - 1:
                wait_time = backoff_factor * (2 ** attempt) * 2  # Longer wait for overload
                print(f"⏳ Service overloaded. Waiting {wait_time} seconds...")
                time.sleep(wait_time)
                continue
            else:
                print("❌ Service overloaded. Max retries reached.")
                return None
                
        except APITimeoutError as e:
            logging.warning(f"Request timeout (attempt {attempt + 1}): {e}")
            if attempt < max_retries - 1:
                print(f"⏳ Request timed out. Retrying...")
                continue
            else:
                print("❌ Request timeout. Max retries reached.")
                return None
                
        except APIConnectionError as e:
            logging.warning(f"Connection error (attempt {attempt + 1}): {e}")
            if attempt < max_retries - 1:
                wait_time = backoff_factor * (2 ** attempt)
                print(f"⏳ Connection error. Retrying in {wait_time} seconds...")
                time.sleep(wait_time)
                continue
            else:
                print("❌ Connection error. Max retries reached.")
                return None
                
        except BadRequestError as e:
            logging.error(f"Bad request: {e}")
            print(f"❌ Bad request: {e}")
            print("Please check your request parameters.")
            return None
            
        except NotFoundError as e:
            logging.error(f"Resource not found: {e}")
            print(f"❌ Resource not found: {e}")
            return None
            
        except UnprocessableEntityError as e:
            logging.error(f"Unprocessable entity: {e}")
            print(f"❌ Request could not be processed: {e}")
            return None
            
        except InternalServerError as e:
            logging.error(f"Internal server error (attempt {attempt + 1}): {e}")
            if attempt < max_retries - 1:
                wait_time = backoff_factor * (2 ** attempt) * 3  # Longer wait for server errors
                print(f"⏳ Server error. Retrying in {wait_time} seconds...")
                time.sleep(wait_time)
                continue
            else:
                print("❌ Server error. Max retries reached.")
                return None
                
        except APIStatusError as e:
            logging.error(f"API status error: {e} (Status: {e.status_code})")
            print(f"❌ API error: {e}")
            return None
            
        except APIError as e:
            logging.error(f"General API error: {e}")
            print(f"❌ API error: {e}")
            return None
            
        except Exception as e:
            logging.error(f"Unexpected error: {e}")
            print(f"❌ Unexpected error: {e}")
            return None
    
    return None

# Usage
def create_message():
    return client.messages.create(
        model="claude-sonnet-4-20250514",
        max_tokens=1024,
        messages=[{"role": "user", "content": "Hello!"}]
    )

result = handle_anthropic_request(create_message, max_retries=3, backoff_factor=1.5)
if result:
    print("✅ Request successful!")
    print(result.content[0].text)

Streaming Error Handling

def safe_streaming_request(client: Anthropic, **kwargs) -> Optional[str]:
    """Handle errors in streaming requests"""
    
    try:
        with client.messages.stream(**kwargs) as stream:
            accumulated_text = ""
            
            try:
                for text in stream.text_stream:
                    accumulated_text += text
                    print(text, end="", flush=True)
                    
                return accumulated_text
                
            except Exception as stream_error:
                print(f"\n❌ Streaming error: {stream_error}")
                
                # Try to get partial result
                try:
                    partial_message = stream.get_final_message()
                    if partial_message and partial_message.content:
                        print("\n⚠️ Returning partial result...")
                        return partial_message.content[0].text
                except:
                    pass
                    
                return accumulated_text if accumulated_text else None
                
    except RateLimitError as e:
        print(f"❌ Rate limit in streaming: {e}")
        return None
        
    except APITimeoutError as e:
        print(f"❌ Timeout in streaming: {e}")
        return None
        
    except APIConnectionError as e:
        print(f"❌ Connection error in streaming: {e}")
        return None
        
    except APIError as e:
        print(f"❌ API error in streaming: {e}")
        return None

# Usage
result = safe_streaming_request(
    client,
    model="claude-sonnet-4-20250514",
    max_tokens=1024,
    messages=[{"role": "user", "content": "Write a short story"}]
)

if result:
    print(f"\n✅ Streaming completed. Total characters: {len(result)}")

Custom Error Context

class AnthropicRequestContext:
    """Context manager for Anthropic requests with enhanced error reporting"""
    
    def __init__(self, operation_name: str, client: Anthropic):
        self.operation_name = operation_name
        self.client = client
        self.start_time = None
        
    def __enter__(self):
        self.start_time = time.time()
        print(f"🚀 Starting {self.operation_name}...")
        return self
        
    def __exit__(self, exc_type, exc_val, exc_tb):
        duration = time.time() - self.start_time if self.start_time else 0
        
        if exc_type is None:
            print(f"✅ {self.operation_name} completed in {duration:.2f}s")
        else:
            print(f"❌ {self.operation_name} failed after {duration:.2f}s")
            
            if issubclass(exc_type, AuthenticationError):
                print("🔑 Authentication issue - check API key")
            elif issubclass(exc_type, RateLimitError):
                print("⏳ Rate limited - consider request pacing")
            elif issubclass(exc_type, APITimeoutError):
                print("⏰ Request timed out - consider shorter requests")
            elif issubclass(exc_type, APIConnectionError):
                print("🌐 Network issue - check internet connection")
            elif issubclass(exc_type, OverloadedError):
                print("🔥 Service overloaded - try again later")
            elif issubclass(exc_type, APIError):
                print(f"🔧 API error: {exc_val}")
            else:
                print(f"💥 Unexpected error: {exc_val}")
        
        # Don't suppress the exception
        return False

# Usage
with AnthropicRequestContext("Message Creation", client) as ctx:
    message = client.messages.create(
        model="claude-sonnet-4-20250514",
        max_tokens=1024,
        messages=[{"role": "user", "content": "Hello!"}]
    )
    print(message.content[0].text)

Error Logging and Monitoring

import json
from datetime import datetime
from pathlib import Path

class AnthropicErrorLogger:
    """Logger for tracking and analyzing Anthropic API errors"""
    
    def __init__(self, log_file: str = "anthropic_errors.log"):
        self.log_file = Path(log_file)
        self.error_counts = {}
        
    def log_error(self, error: Exception, context: dict = None):
        """Log error with context information"""
        
        error_info = {
            "timestamp": datetime.now().isoformat(),
            "error_type": type(error).__name__,
            "error_message": str(error),
            "context": context or {}
        }
        
        # Add specific error details
        if isinstance(error, APIStatusError):
            error_info["status_code"] = error.status_code
            error_info["response_body"] = str(error.body) if error.body else None
            
        # Track error frequency
        error_key = type(error).__name__
        self.error_counts[error_key] = self.error_counts.get(error_key, 0) + 1
        
        # Write to log file
        with open(self.log_file, "a") as f:
            f.write(json.dumps(error_info) + "\n")
            
        print(f"📝 Error logged: {error_key}")
        
    def get_error_summary(self) -> dict:
        """Get summary of logged errors"""
        return {
            "total_errors": sum(self.error_counts.values()),
            "error_counts": self.error_counts,
            "most_common": max(self.error_counts.items(), key=lambda x: x[1]) if self.error_counts else None
        }

# Usage
error_logger = AnthropicErrorLogger()

def logged_request(request_func, context: dict = None):
    """Execute request with error logging"""
    try:
        return request_func()
    except Exception as e:
        error_logger.log_error(e, context)
        raise

# Example usage
try:
    result = logged_request(
        lambda: client.messages.create(
            model="claude-sonnet-4-20250514",
            max_tokens=1024,
            messages=[{"role": "user", "content": "Hello!"}]
        ),
        context={"user_id": "user123", "request_type": "chat"}
    )
except Exception:
    pass

# Get error summary
summary = error_logger.get_error_summary()
print(f"Error summary: {summary}")

Async Error Handling

import asyncio
from anthropic import AsyncAnthropic

async def async_error_handler():
    """Demonstrate async error handling patterns"""
    
    client = AsyncAnthropic()
    
    try:
        message = await client.messages.create(
            model="claude-sonnet-4-20250514",
            max_tokens=1024,
            messages=[{"role": "user", "content": "Hello async world!"}]
        )
        
        return message.content[0].text
        
    except RateLimitError as e:
        print(f"Rate limited in async: {e}")
        # Could implement async backoff here
        await asyncio.sleep(5)
        return None
        
    except APITimeoutError as e:
        print(f"Async timeout: {e}")
        return None
        
    except APIError as e:
        print(f"Async API error: {e}")
        return None

# Usage
async def main():
    result = await async_error_handler()
    if result:
        print(f"Async result: {result}")

asyncio.run(main())

Install with Tessl CLI