/ error_handling.py
error_handling.py
1 """ 2 Comprehensive error handling framework for Model Control Protocol (MCP) systems. 3 4 This module implements a consistent, standardized approach to error handling for MCP tools 5 and services. It provides decorators, formatters, and utilities that transform Python 6 exceptions into structured, protocol-compliant error responses that LLMs and client 7 applications can reliably interpret and respond to. 8 9 The framework is designed around several key principles: 10 11 1. CONSISTENCY: All errors follow the same structured format regardless of their source 12 2. RECOVERABILITY: Errors include explicit information on whether operations can be retried 13 3. ACTIONABILITY: Error responses provide specific suggestions for resolving issues 14 4. DEBUGGABILITY: Rich error details are preserved for troubleshooting 15 5. CATEGORIZATION: Errors are mapped to standardized types for consistent handling 16 17 Key components: 18 - ErrorType enum: Categorization system for different error conditions 19 - format_error_response(): Creates standardized error response dictionaries 20 - with_error_handling: Decorator that catches exceptions and formats responses 21 - validate_inputs: Decorator for declarative parameter validation 22 - Validator functions: Reusable validation logic for common parameter types 23 24 Usage example: 25 ```python 26 @with_error_handling 27 @validate_inputs( 28 prompt=non_empty_string, 29 temperature=in_range(0.0, 1.0) 30 ) 31 async def generate_text(prompt, temperature=0.7): 32 # Implementation... 33 # Any exceptions thrown here will be caught and formatted 34 # Input validation happens before execution 35 if external_service_down: 36 raise Exception("External service unavailable") 37 return result 38 ``` 39 40 The error handling pattern is designed to work seamlessly with async functions and 41 integrates with the MCP protocol's expected error response structure. 42 """ 43 44 import functools 45 import inspect 46 import time 47 import traceback 48 from enum import Enum 49 from typing import Any, Callable, Dict, Optional, Union 50 51 52 class ErrorType(str, Enum): 53 """Types of errors that can occur in MCP tools.""" 54 55 VALIDATION_ERROR = "validation_error" # Input validation failed 56 EXECUTION_ERROR = "execution_error" # Error during execution 57 PERMISSION_ERROR = "permission_error" # Insufficient permissions 58 NOT_FOUND_ERROR = "not_found_error" # Resource not found 59 TIMEOUT_ERROR = "timeout_error" # Operation timed out 60 RATE_LIMIT_ERROR = "rate_limit_error" # Rate limit exceeded 61 EXTERNAL_ERROR = "external_error" # Error in external service 62 UNKNOWN_ERROR = "unknown_error" # Unknown error 63 64 65 def format_error_response( 66 error_type: Union[ErrorType, str], 67 message: str, 68 details: Optional[Dict[str, Any]] = None, 69 retriable: bool = False, 70 suggestions: Optional[list] = None, 71 ) -> Dict[str, Any]: 72 """ 73 Format a standardized error response. 74 75 Args: 76 error_type: Type of error 77 message: Human-readable error message 78 details: Additional error details 79 retriable: Whether the operation can be retried 80 suggestions: List of suggestions for resolving the error 81 82 Returns: 83 Formatted error response 84 """ 85 return { 86 "success": False, 87 "isError": True, # MCP protocol flag 88 "error": { 89 "type": error_type if isinstance(error_type, str) else error_type.value, 90 "message": message, 91 "details": details or {}, 92 "retriable": retriable, 93 "suggestions": suggestions or [], 94 "timestamp": time.time(), 95 }, 96 } 97 98 99 def with_error_handling(func: Callable) -> Callable: 100 """ 101 Decorator that provides standardized exception handling for MCP tool functions. 102 103 This decorator intercepts any exceptions raised by the wrapped function and transforms 104 them into a structured error response format that follows the MCP protocol. The response 105 includes consistent error categorization, helpful suggestions for recovery, and details 106 to aid debugging. 107 108 Key features: 109 - Automatically categorizes exceptions into appropriate ErrorType values 110 - Preserves the original exception message and stack trace 111 - Adds relevant suggestions based on the error type 112 - Indicates whether the operation can be retried 113 - Adds a timestamp for error logging/tracking 114 115 The error response structure always includes: 116 - success: False 117 - isError: True (MCP protocol flag) 118 - error: A dictionary with type, message, details, retriable flag, and suggestions 119 120 Exception mapping: 121 - ValueError, TypeError, KeyError, AttributeError → VALIDATION_ERROR (retriable) 122 - FileNotFoundError, KeyError, IndexError → NOT_FOUND_ERROR (not retriable) 123 - PermissionError, AccessError → PERMISSION_ERROR (not retriable) 124 - TimeoutError → TIMEOUT_ERROR (retriable) 125 - Exceptions with "rate limit" in message → RATE_LIMIT_ERROR (retriable) 126 - All other exceptions → UNKNOWN_ERROR (not retriable) 127 128 Args: 129 func: The async function to wrap with error handling 130 131 Returns: 132 Decorated async function that catches exceptions and returns structured error responses 133 134 Example: 135 ```python 136 @with_error_handling 137 async def my_tool_function(param1, param2): 138 # Function implementation that might raise exceptions 139 # If an exception occurs, it will be transformed into a structured response 140 ``` 141 """ 142 143 @functools.wraps(func) 144 async def wrapper(*args, **kwargs): 145 try: 146 # Call the original function 147 return await func(*args, **kwargs) 148 except Exception as e: 149 # Get exception details 150 exc_type = type(e).__name__ 151 exc_message = str(e) 152 exc_traceback = traceback.format_exc() 153 154 # Determine error type 155 error_type = ErrorType.UNKNOWN_ERROR 156 retriable = False 157 158 # Map common exceptions to error types 159 if exc_type in ("ValueError", "TypeError", "KeyError", "AttributeError"): 160 error_type = ErrorType.VALIDATION_ERROR 161 retriable = True 162 elif exc_type in ("FileNotFoundError", "KeyError", "IndexError"): 163 error_type = ErrorType.NOT_FOUND_ERROR 164 retriable = False 165 elif exc_type in ("PermissionError", "AccessError"): 166 error_type = ErrorType.PERMISSION_ERROR 167 retriable = False 168 elif exc_type in ("TimeoutError"): 169 error_type = ErrorType.TIMEOUT_ERROR 170 retriable = True 171 elif "rate limit" in exc_message.lower(): 172 error_type = ErrorType.RATE_LIMIT_ERROR 173 retriable = True 174 175 # Generate suggestions based on error type 176 suggestions = [] 177 if error_type == ErrorType.VALIDATION_ERROR: 178 suggestions = [ 179 "Check that all required parameters are provided", 180 "Verify parameter types and formats", 181 "Ensure parameter values are within allowed ranges", 182 ] 183 elif error_type == ErrorType.NOT_FOUND_ERROR: 184 suggestions = [ 185 "Verify the resource ID or path exists", 186 "Check for typos in identifiers", 187 "Ensure the resource hasn't been deleted", 188 ] 189 elif error_type == ErrorType.RATE_LIMIT_ERROR: 190 suggestions = [ 191 "Wait before retrying the request", 192 "Reduce the frequency of requests", 193 "Implement backoff strategy for retries", 194 ] 195 196 # Format and return error response 197 return format_error_response( 198 error_type=error_type, 199 message=exc_message, 200 details={"exception_type": exc_type, "traceback": exc_traceback}, 201 retriable=retriable, 202 suggestions=suggestions, 203 ) 204 205 return wrapper 206 207 208 def validate_inputs(**validators): 209 """ 210 Decorator for validating tool input parameters against custom validation rules. 211 212 This decorator enables declarative input validation for async tool functions by applying 213 validator functions to specified parameters before the decorated function is called. 214 If any validation fails, the function returns a standardized error response instead 215 of executing, preventing errors from propagating and providing clear feedback on the issue. 216 217 The validation approach supports: 218 - Applying different validation rules to different parameters 219 - Detailed error messages explaining which parameter failed and why 220 - Custom validation logic via any callable that raises ValueError on failure 221 - Zero validation overhead for parameters not explicitly validated 222 223 Validator functions should: 224 1. Take a single parameter (the value to validate) 225 2. Raise a ValueError with a descriptive message if validation fails 226 3. Return None or any value (which is ignored) if validation passes 227 4. Include a docstring that describes the constraint (used in error messages) 228 229 Args: 230 **validators: A mapping of parameter names to validator functions. 231 Each key should match a parameter name in the decorated function. 232 Each value should be a callable that validates the corresponding parameter. 233 234 Returns: 235 Decorator function that wraps an async function with input validation 236 237 Example: 238 ``` 239 # Define validators (or use the provided ones like non_empty_string) 240 def validate_temperature(value): 241 '''Temperature must be between 0.0 and 1.0.''' 242 if not isinstance(value, float) or value < 0.0 or value > 1.0: 243 raise ValueError("Temperature must be between 0.0 and 1.0") 244 245 # Apply validation to specific parameters 246 @validate_inputs( 247 prompt=non_empty_string, 248 temperature=validate_temperature, 249 max_tokens=positive_number 250 ) 251 async def generate_text(prompt, temperature=0.7, max_tokens=None): 252 # This function will only be called if all validations pass 253 # Otherwise a standardized error response is returned 254 ... 255 256 # The response structure when validation fails: 257 # { 258 # "success": False, 259 # "isError": True, 260 # "error": { 261 # "type": "validation_error", 262 # "message": "Invalid value for parameter 'prompt': Value must be a non-empty string", 263 # "details": { ... }, 264 # "retriable": true, 265 # "suggestions": [ ... ] 266 # } 267 # } 268 ``` 269 270 Note: 271 This decorator should typically be applied before other decorators like 272 with_error_handling so that validation errors are correctly formatted. 273 """ 274 275 def decorator(func): 276 @functools.wraps(func) 277 async def wrapper(*args, **kwargs): 278 # Get function signature 279 sig = inspect.signature(func) 280 281 # Build mapping of parameter names to values 282 bound_args = sig.bind(*args, **kwargs) 283 bound_args.apply_defaults() 284 285 # Validate inputs 286 for param_name, validator in validators.items(): 287 if param_name in bound_args.arguments: 288 value = bound_args.arguments[param_name] 289 try: 290 # Run validation 291 validator(value) 292 except Exception as e: 293 # Return validation error 294 return format_error_response( 295 error_type=ErrorType.VALIDATION_ERROR, 296 message=f"Invalid value for parameter '{param_name}': {str(e)}", 297 details={ 298 "parameter": param_name, 299 "value": str(value), 300 "constraint": str(validator.__doc__ or ""), 301 }, 302 retriable=True, 303 suggestions=[ 304 f"Provide a valid value for '{param_name}'", 305 "Check the parameter constraints in the tool description", 306 ], 307 ) 308 309 # Call the original function if validation passes 310 return await func(*args, **kwargs) 311 312 return wrapper 313 314 return decorator 315 316 317 # Example validators 318 def non_empty_string(value): 319 """ 320 Validates that a value is a non-empty string. 321 322 This validator checks that the input is a string type and contains at least 323 one non-whitespace character. Empty strings or strings containing only 324 whitespace characters are rejected. This is useful for validating required 325 text inputs where blank values should not be allowed. 326 327 Args: 328 value: The value to validate 329 330 Raises: 331 ValueError: If the value is not a string or is empty/whitespace-only 332 """ 333 if not isinstance(value, str) or not value.strip(): 334 raise ValueError("Value must be a non-empty string") 335 336 337 def positive_number(value): 338 """ 339 Validates that a value is a positive number (greater than zero). 340 341 This validator ensures that the input is either an integer or float 342 and has a value greater than zero. Zero or negative values are rejected. 343 This is useful for validating inputs like quantities, counts, or rates 344 that must be positive. 345 346 Args: 347 value: The value to validate 348 349 Raises: 350 ValueError: If the value is not a number or is not positive 351 """ 352 if not isinstance(value, (int, float)) or value <= 0: 353 raise ValueError("Value must be a positive number") 354 355 356 def in_range(min_val, max_val): 357 """ 358 Creates a validator function for checking if a number falls within a specified range. 359 360 This is a validator factory that returns a custom validator function 361 configured with the given minimum and maximum bounds. The returned function 362 checks that a value is a number and falls within the inclusive range 363 [min_val, max_val]. This is useful for validating inputs that must fall 364 within specific limits, such as probabilities, temperatures, or indexes. 365 366 Args: 367 min_val: The minimum allowed value (inclusive) 368 max_val: The maximum allowed value (inclusive) 369 370 Returns: 371 A validator function that checks if values are within the specified range 372 373 Example: 374 ```python 375 # Create a validator for temperature (0.0 to 1.0) 376 validate_temperature = in_range(0.0, 1.0) 377 378 # Use in validation decorator 379 @validate_inputs(temperature=validate_temperature) 380 async def generate_text(prompt, temperature=0.7): 381 # Function body 382 ... 383 ``` 384 """ 385 386 def validator(value): 387 """Value must be between {min_val} and {max_val}.""" 388 if not isinstance(value, (int, float)) or value < min_val or value > max_val: 389 raise ValueError(f"Value must be between {min_val} and {max_val}") 390 391 return validator