CtrlK
BlogDocsLog inGet started
Tessl Logo

tessl/pypi-anthropic

The official Python library for the anthropic API

Pending
Quality

Pending

Does it follow best practices?

Impact

Pending

No eval scenarios have been run

SecuritybySnyk

Pending

The risk profile of this skill

Overview
Eval results
Files

errors.mddocs/

Error Handling

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())

docs

batching.md

bedrock.md

beta.md

completions.md

configuration.md

errors.md

index.md

messages.md

models.md

streaming.md

tools.md

vertex.md

tile.json