Zhiwen Assistant MCP

""" 错误处理中间件 提供统一的错误处理、错误分类和用户友好的错误信息。 """ import traceback from typing import Optional, Dict, Any, Type from enum import Enum import logging from .logging import logging_middleware class ErrorSeverity(Enum): """错误严重程度""" LOW = "low" # 轻微错误，不影响主要功能 MEDIUM = "medium" # 中等错误，影响部分功能 HIGH = "high" # 严重错误，影响核心功能 CRITICAL = "critical" # 致命错误，系统无法正常运行 class ErrorCategory(Enum): """错误类别""" VALIDATION = "validation" # 参数验证错误 PERMISSION = "permission" # 权限错误 FILE_SYSTEM = "file_system" # 文件系统错误 NETWORK = "network" # 网络错误 CONFIGURATION = "configuration" # 配置错误 PERFORMANCE = "performance" # 性能错误 UNKNOWN = "unknown" # 未知错误 class MCPServerError(Exception): """MCP服务器自定义错误基类""" def __init__( self, message: str, category: ErrorCategory = ErrorCategory.UNKNOWN, severity: ErrorSeverity = ErrorSeverity.MEDIUM, details: Optional[Dict[str, Any]] = None, suggestions: Optional[str] = None ): self.message = message self.category = category self.severity = severity self.details = details or {} self.suggestions = suggestions super().__init__(self.message) class ValidationError(MCPServerError): """参数验证错误""" def __init__(self, message: str, field: str = None, value: Any = None, **kwargs): details = kwargs.get('details', {}) if field: details['field'] = field if value is not None: details['invalid_value'] = str(value) super().__init__( message=message, category=ErrorCategory.VALIDATION, severity=ErrorSeverity.LOW, details=details, **kwargs ) class PermissionError(MCPServerError): """权限错误""" def __init__(self, message: str, required_permission: str = None, **kwargs): details = kwargs.get('details', {}) if required_permission: details['required_permission'] = required_permission super().__init__( message=message, category=ErrorCategory.PERMISSION, severity=ErrorSeverity.HIGH, details=details, **kwargs ) class FileSystemError(MCPServerError): """文件系统错误""" def __init__(self, message: str, path: str = None, operation: str = None, **kwargs): details = kwargs.get('details', {}) if path: details['path'] = path if operation: details['operation'] = operation super().__init__( message=message, category=ErrorCategory.FILE_SYSTEM, severity=ErrorSeverity.MEDIUM, details=details, **kwargs ) class ConfigurationError(MCPServerError): """配置错误""" def __init__(self, message: str, config_key: str = None, **kwargs): details = kwargs.get('details', {}) if config_key: details['config_key'] = config_key super().__init__( message=message, category=ErrorCategory.CONFIGURATION, severity=ErrorSeverity.HIGH, details=details, **kwargs ) class ErrorHandlerMiddleware: """错误处理中间件""" def __init__(self): self.error_mapping = { ValueError: ValidationError, PermissionError: PermissionError, FileNotFoundError: FileSystemError, NotADirectoryError: FileSystemError, IOError: FileSystemError, OSError: FileSystemError, } def categorize_error(self, error: Exception) -> ErrorCategory: """根据错误类型自动分类""" error_type = type(error).__name__.lower() if any(keyword in error_type for keyword in ['value', 'validation', 'type']): return ErrorCategory.VALIDATION elif any(keyword in error_type for keyword in ['permission', 'access', 'auth']): return ErrorCategory.PERMISSION elif any(keyword in error_type for keyword in ['file', 'path', 'directory', 'io']): return ErrorCategory.FILE_SYSTEM elif any(keyword in error_type for keyword in ['network', 'connection', 'timeout']): return ErrorCategory.NETWORK elif any(keyword in error_type for keyword in ['config', 'setting']): return ErrorCategory.CONFIGURATION elif any(keyword in error_type for keyword in ['memory', 'performance', 'timeout']): return ErrorCategory.PERFORMANCE else: return ErrorCategory.UNKNOWN def determine_severity(self, error: Exception) -> ErrorSeverity: """根据错误类型确定严重程度""" if isinstance(error, (ValidationError,)): return ErrorSeverity.LOW elif isinstance(error, (FileSystemError, ConfigurationError)): return ErrorSeverity.MEDIUM elif isinstance(error, (PermissionError,)): return ErrorSeverity.HIGH else: return ErrorSeverity.MEDIUM def get_suggestion(self, error: Exception) -> str: """生成错误解决建议""" if isinstance(error, ValidationError): return "请检查输入参数的格式和值是否符合要求" elif isinstance(error, PermissionError): return "请检查文件路径是否正确，以及是否有足够的访问权限" elif isinstance(error, FileSystemError): return "请检查文件路径是否存在，以及是否有读写权限" elif isinstance(error, ConfigurationError): return "请检查配置文件是否正确，或者使用默认配置" else: return "请查看详细错误信息，如有需要请联系技术支持" def handle_error(self, error: Exception, tool_name: str = "", context: str = "") -> MCPServerError: """ 统一处理错误 Args: error: 原始异常 tool_name: 调用的工具名称 context: 错误上下文信息 Returns: 处理后的MCP服务器错误 """ # 如果已经是我们的自定义错误，直接返回 if isinstance(error, MCPServerError): mcp_error = error else: # 将系统错误转换为我们的自定义错误 mcp_error_class = self.error_mapping.get(type(error), MCPServerError) category = self.categorize_error(error) severity = self.determine_severity(error) suggestions = self.generate_suggestions(error) details = { 'original_error_type': type(error).__name__, 'original_error_message': str(error) } if context: details.update(context) mcp_error = mcp_error_class( message=str(error), category=category, severity=severity, details=details, suggestions=suggestions ) # 记录错误日志 logging_middleware.log_error( tool_name or "unknown", error, context or {} ) return mcp_error def format_error_response(self, error: MCPServerError) -> Dict[str, Any]: """ 格式化错误响应 Args: error: MCP服务器错误 Returns: 格式化的错误响应 """ return { "success": False, "error": { "type": error.__class__.__name__, "message": error.message, "category": error.category.value, "severity": error.severity.value, "details": error.details, "suggestions": error.suggestions } } # 全局错误处理中间件实例 error_handler_middleware = ErrorHandlerMiddleware() def handle_errors(tool_name: str = None): """错误处理装饰器""" def decorator(func): def wrapper(*args, **kwargs): try: return func(*args, **kwargs) except (RuntimeError, ValueError) as e: # 构建错误上下文 context = { 'function_name': func.__name__, 'args_count': len(args), 'kwargs_keys': list(kwargs.keys()) } # 处理错误 mcp_error = error_handler_middleware.handle_error( e, tool_name or func.__name__, context ) # 返回格式化的错误响应 return error_handler_middleware.format_error_response(mcp_error) async def async_wrapper(*args, **kwargs): try: return await func(*args, **kwargs) except (RuntimeError, ValueError) as e: # 构建错误上下文 context = { 'function_name': func.__name__, 'args_count': len(args), 'kwargs_keys': list(kwargs.keys()) } # 处理错误 mcp_error = error_handler_middleware.handle_error( e, tool_name or func.__name__, context ) # 返回格式化的错误响应 return error_handler_middleware.format_error_response(mcp_error) import asyncio if asyncio.iscoroutinefunction(func): return async_wrapper else: return wrapper return decorator