pypi-anthropic

1
# Error Handling
2

3
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.
4

5
## Capabilities
6

7
### Exception Hierarchy
8

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

11
```python { .api }
12
class AnthropicError(Exception):
13
    """Base exception for all Anthropic SDK errors"""
14
    pass
15

16
class APIError(AnthropicError):
17
    """Base class for API-related errors"""
18
    pass
19

20
class APIStatusError(APIError):
21
    """HTTP status code errors"""
22
    def __init__(self, message: str, *, response: httpx.Response, body: Optional[object]): ...
23
    
24
    @property
25
    def status_code(self) -> int: ...
26
    @property
27
    def response(self) -> httpx.Response: ...
28
    @property  
29
    def body(self) -> Optional[object]: ...
30

31
class APITimeoutError(APIError):
32
    """Request timeout errors"""
33
    pass
34

35
class APIConnectionError(APIError):
36
    """Network connection errors"""
37
    pass
38

39
class APIResponseValidationError(APIError):
40
    """Response validation errors"""
41
    pass
42
```
43

44
### HTTP Status Errors
45

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

48
```python { .api }
49
class BadRequestError(APIStatusError):
50
    """HTTP 400 Bad Request errors"""
51
    pass
52

53
class AuthenticationError(APIStatusError):
54
    """HTTP 401 Authentication errors"""
55
    pass
56

57
class PermissionDeniedError(APIStatusError):
58
    """HTTP 403 Permission Denied errors"""
59
    pass
60

61
class NotFoundError(APIStatusError):
62
    """HTTP 404 Not Found errors"""
63
    pass
64

65
class ConflictError(APIStatusError):
66
    """HTTP 409 Conflict errors"""
67
    pass
68

69
class UnprocessableEntityError(APIStatusError):
70
    """HTTP 422 Unprocessable Entity errors"""
71
    pass
72

73
class RateLimitError(APIStatusError):
74
    """HTTP 429 Rate Limit errors"""
75
    pass
76

77
class InternalServerError(APIStatusError):
78
    """HTTP 500 Internal Server errors"""
79
    pass
80
```
81

82
### Error Response Types
83

84
Structured error response objects that provide detailed error information.
85

86
```python { .api }
87
class ErrorObject(TypedDict):
88
    type: str
89
    message: str
90

91
class ErrorResponse(TypedDict):
92
    type: Literal["error"]
93
    error: ErrorObject
94

95
class APIErrorObject(TypedDict):
96
    type: str
97
    message: str
98

99
class AuthenticationError(TypedDict):
100
    type: Literal["authentication_error"]
101
    message: str
102

103
class PermissionError(TypedDict):
104
    type: Literal["permission_error"]
105
    message: str
106

107
class NotFoundError(TypedDict):
108
    type: Literal["not_found_error"]
109
    message: str
110

111
class RateLimitError(TypedDict):
112
    type: Literal["rate_limit_error"]
113
    message: str
114

115
class OverloadedError(TypedDict):
116
    type: Literal["overloaded_error"]
117
    message: str
118

119
class BillingError(TypedDict):
120
    type: Literal["billing_error"]
121
    message: str
122

123
class GatewayTimeoutError(TypedDict):
124
    type: Literal["gateway_timeout_error"]
125
    message: str
126

127
class InvalidRequestError(TypedDict):
128
    type: Literal["invalid_request_error"]
129
    message: str
130
```
131

132
## Usage Examples
133

134
### Basic Error Handling
135

136
```python
137
from anthropic import Anthropic, APIError, RateLimitError, AuthenticationError
138

139
client = Anthropic()
140

141
try:
142
    message = client.messages.create(
143
        model="claude-sonnet-4-20250514",
144
        max_tokens=1024,
145
        messages=[
146
            {"role": "user", "content": "Hello!"}
147
        ]
148
    )
149
    print(message.content[0].text)
150
    
151
except AuthenticationError as e:
152
    print(f"Authentication failed: {e}")
153
    print("Please check your API key")
154
    
155
except RateLimitError as e:
156
    print(f"Rate limit exceeded: {e}")
157
    print("Please wait before making more requests")
158
    
159
except APIError as e:
160
    print(f"API error occurred: {e}")
161
    print(f"Status code: {e.status_code}")
162
    
163
except Exception as e:
164
    print(f"Unexpected error: {e}")
165
```
166

167
### Comprehensive Error Handler
168

169
```python
170
import time
171
import logging
172
from typing import Optional, Any
173

174
def handle_anthropic_request(
175
    request_func: callable,
176
    *args,
177
    max_retries: int = 3,
178
    backoff_factor: float = 1.0,
179
    **kwargs
180
) -> Optional[Any]:
181
    """
182
    Execute an Anthropic API request with comprehensive error handling and retries
183
    """
184
    
185
    for attempt in range(max_retries):
186
        try:
187
            return request_func(*args, **kwargs)
188
            
189
        except AuthenticationError as e:
190
            logging.error(f"Authentication error: {e}")
191
            print("❌ Authentication failed. Please check your API key.")
192
            return None
193
            
194
        except PermissionDeniedError as e:
195
            logging.error(f"Permission denied: {e}")
196
            print("❌ Permission denied. Check your account permissions.")
197
            return None
198
            
199
        except RateLimitError as e:
200
            logging.warning(f"Rate limit exceeded (attempt {attempt + 1}): {e}")
201
            if attempt < max_retries - 1:
202
                wait_time = backoff_factor * (2 ** attempt)
203
                print(f"⏳ Rate limited. Waiting {wait_time} seconds...")
204
                time.sleep(wait_time)
205
                continue
206
            else:
207
                print("❌ Rate limit exceeded. Max retries reached.")
208
                return None
209
                
210
        except OverloadedError as e:
211
            logging.warning(f"Service overloaded (attempt {attempt + 1}): {e}")
212
            if attempt < max_retries - 1:
213
                wait_time = backoff_factor * (2 ** attempt) * 2  # Longer wait for overload
214
                print(f"⏳ Service overloaded. Waiting {wait_time} seconds...")
215
                time.sleep(wait_time)
216
                continue
217
            else:
218
                print("❌ Service overloaded. Max retries reached.")
219
                return None
220
                
221
        except APITimeoutError as e:
222
            logging.warning(f"Request timeout (attempt {attempt + 1}): {e}")
223
            if attempt < max_retries - 1:
224
                print(f"⏳ Request timed out. Retrying...")
225
                continue
226
            else:
227
                print("❌ Request timeout. Max retries reached.")
228
                return None
229
                
230
        except APIConnectionError as e:
231
            logging.warning(f"Connection error (attempt {attempt + 1}): {e}")
232
            if attempt < max_retries - 1:
233
                wait_time = backoff_factor * (2 ** attempt)
234
                print(f"⏳ Connection error. Retrying in {wait_time} seconds...")
235
                time.sleep(wait_time)
236
                continue
237
            else:
238
                print("❌ Connection error. Max retries reached.")
239
                return None
240
                
241
        except BadRequestError as e:
242
            logging.error(f"Bad request: {e}")
243
            print(f"❌ Bad request: {e}")
244
            print("Please check your request parameters.")
245
            return None
246
            
247
        except NotFoundError as e:
248
            logging.error(f"Resource not found: {e}")
249
            print(f"❌ Resource not found: {e}")
250
            return None
251
            
252
        except UnprocessableEntityError as e:
253
            logging.error(f"Unprocessable entity: {e}")
254
            print(f"❌ Request could not be processed: {e}")
255
            return None
256
            
257
        except InternalServerError as e:
258
            logging.error(f"Internal server error (attempt {attempt + 1}): {e}")
259
            if attempt < max_retries - 1:
260
                wait_time = backoff_factor * (2 ** attempt) * 3  # Longer wait for server errors
261
                print(f"⏳ Server error. Retrying in {wait_time} seconds...")
262
                time.sleep(wait_time)
263
                continue
264
            else:
265
                print("❌ Server error. Max retries reached.")
266
                return None
267
                
268
        except APIStatusError as e:
269
            logging.error(f"API status error: {e} (Status: {e.status_code})")
270
            print(f"❌ API error: {e}")
271
            return None
272
            
273
        except APIError as e:
274
            logging.error(f"General API error: {e}")
275
            print(f"❌ API error: {e}")
276
            return None
277
            
278
        except Exception as e:
279
            logging.error(f"Unexpected error: {e}")
280
            print(f"❌ Unexpected error: {e}")
281
            return None
282
    
283
    return None
284

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

293
result = handle_anthropic_request(create_message, max_retries=3, backoff_factor=1.5)
294
if result:
295
    print("✅ Request successful!")
296
    print(result.content[0].text)
297
```
298

299
### Streaming Error Handling
300

301
```python
302
def safe_streaming_request(client: Anthropic, **kwargs) -> Optional[str]:
303
    """Handle errors in streaming requests"""
304
    
305
    try:
306
        with client.messages.stream(**kwargs) as stream:
307
            accumulated_text = ""
308
            
309
            try:
310
                for text in stream.text_stream:
311
                    accumulated_text += text
312
                    print(text, end="", flush=True)
313
                    
314
                return accumulated_text
315
                
316
            except Exception as stream_error:
317
                print(f"\n❌ Streaming error: {stream_error}")
318
                
319
                # Try to get partial result
320
                try:
321
                    partial_message = stream.get_final_message()
322
                    if partial_message and partial_message.content:
323
                        print("\n⚠️ Returning partial result...")
324
                        return partial_message.content[0].text
325
                except:
326
                    pass
327
                    
328
                return accumulated_text if accumulated_text else None
329
                
330
    except RateLimitError as e:
331
        print(f"❌ Rate limit in streaming: {e}")
332
        return None
333
        
334
    except APITimeoutError as e:
335
        print(f"❌ Timeout in streaming: {e}")
336
        return None
337
        
338
    except APIConnectionError as e:
339
        print(f"❌ Connection error in streaming: {e}")
340
        return None
341
        
342
    except APIError as e:
343
        print(f"❌ API error in streaming: {e}")
344
        return None
345

346
# Usage
347
result = safe_streaming_request(
348
    client,
349
    model="claude-sonnet-4-20250514",
350
    max_tokens=1024,
351
    messages=[{"role": "user", "content": "Write a short story"}]
352
)
353

354
if result:
355
    print(f"\n✅ Streaming completed. Total characters: {len(result)}")
356
```
357

358
### Custom Error Context
359

360
```python
361
class AnthropicRequestContext:
362
    """Context manager for Anthropic requests with enhanced error reporting"""
363
    
364
    def __init__(self, operation_name: str, client: Anthropic):
365
        self.operation_name = operation_name
366
        self.client = client
367
        self.start_time = None
368
        
369
    def __enter__(self):
370
        self.start_time = time.time()
371
        print(f"🚀 Starting {self.operation_name}...")
372
        return self
373
        
374
    def __exit__(self, exc_type, exc_val, exc_tb):
375
        duration = time.time() - self.start_time if self.start_time else 0
376
        
377
        if exc_type is None:
378
            print(f"✅ {self.operation_name} completed in {duration:.2f}s")
379
        else:
380
            print(f"❌ {self.operation_name} failed after {duration:.2f}s")
381
            
382
            if issubclass(exc_type, AuthenticationError):
383
                print("🔑 Authentication issue - check API key")
384
            elif issubclass(exc_type, RateLimitError):
385
                print("⏳ Rate limited - consider request pacing")
386
            elif issubclass(exc_type, APITimeoutError):
387
                print("⏰ Request timed out - consider shorter requests")
388
            elif issubclass(exc_type, APIConnectionError):
389
                print("🌐 Network issue - check internet connection")
390
            elif issubclass(exc_type, OverloadedError):
391
                print("🔥 Service overloaded - try again later")
392
            elif issubclass(exc_type, APIError):
393
                print(f"🔧 API error: {exc_val}")
394
            else:
395
                print(f"💥 Unexpected error: {exc_val}")
396
        
397
        # Don't suppress the exception
398
        return False
399

400
# Usage
401
with AnthropicRequestContext("Message Creation", client) as ctx:
402
    message = client.messages.create(
403
        model="claude-sonnet-4-20250514",
404
        max_tokens=1024,
405
        messages=[{"role": "user", "content": "Hello!"}]
406
    )
407
    print(message.content[0].text)
408
```
409

410
### Error Logging and Monitoring
411

412
```python
413
import json
414
from datetime import datetime
415
from pathlib import Path
416

417
class AnthropicErrorLogger:
418
    """Logger for tracking and analyzing Anthropic API errors"""
419
    
420
    def __init__(self, log_file: str = "anthropic_errors.log"):
421
        self.log_file = Path(log_file)
422
        self.error_counts = {}
423
        
424
    def log_error(self, error: Exception, context: dict = None):
425
        """Log error with context information"""
426
        
427
        error_info = {
428
            "timestamp": datetime.now().isoformat(),
429
            "error_type": type(error).__name__,
430
            "error_message": str(error),
431
            "context": context or {}
432
        }
433
        
434
        # Add specific error details
435
        if isinstance(error, APIStatusError):
436
            error_info["status_code"] = error.status_code
437
            error_info["response_body"] = str(error.body) if error.body else None
438
            
439
        # Track error frequency
440
        error_key = type(error).__name__
441
        self.error_counts[error_key] = self.error_counts.get(error_key, 0) + 1
442
        
443
        # Write to log file
444
        with open(self.log_file, "a") as f:
445
            f.write(json.dumps(error_info) + "\n")
446
            
447
        print(f"📝 Error logged: {error_key}")
448
        
449
    def get_error_summary(self) -> dict:
450
        """Get summary of logged errors"""
451
        return {
452
            "total_errors": sum(self.error_counts.values()),
453
            "error_counts": self.error_counts,
454
            "most_common": max(self.error_counts.items(), key=lambda x: x[1]) if self.error_counts else None
455
        }
456

457
# Usage
458
error_logger = AnthropicErrorLogger()
459

460
def logged_request(request_func, context: dict = None):
461
    """Execute request with error logging"""
462
    try:
463
        return request_func()
464
    except Exception as e:
465
        error_logger.log_error(e, context)
466
        raise
467

468
# Example usage
469
try:
470
    result = logged_request(
471
        lambda: client.messages.create(
472
            model="claude-sonnet-4-20250514",
473
            max_tokens=1024,
474
            messages=[{"role": "user", "content": "Hello!"}]
475
        ),
476
        context={"user_id": "user123", "request_type": "chat"}
477
    )
478
except Exception:
479
    pass
480

481
# Get error summary
482
summary = error_logger.get_error_summary()
483
print(f"Error summary: {summary}")
484
```
485

486
### Async Error Handling
487

488
```python
489
import asyncio
490
from anthropic import AsyncAnthropic
491

492
async def async_error_handler():
493
    """Demonstrate async error handling patterns"""
494
    
495
    client = AsyncAnthropic()
496
    
497
    try:
498
        message = await client.messages.create(
499
            model="claude-sonnet-4-20250514",
500
            max_tokens=1024,
501
            messages=[{"role": "user", "content": "Hello async world!"}]
502
        )
503
        
504
        return message.content[0].text
505
        
506
    except RateLimitError as e:
507
        print(f"Rate limited in async: {e}")
508
        # Could implement async backoff here
509
        await asyncio.sleep(5)
510
        return None
511
        
512
    except APITimeoutError as e:
513
        print(f"Async timeout: {e}")
514
        return None
515
        
516
    except APIError as e:
517
        print(f"Async API error: {e}")
518
        return None
519

520
# Usage
521
async def main():
522
    result = await async_error_handler()
523
    if result:
524
        print(f"Async result: {result}")
525

526
asyncio.run(main())
527
```