src/errors.py

"""
Custom exceptions and error handling for EcoMCP
"""

from typing import Optional, Dict, Any
from src.constants import ErrorCode


class EcoMCPException(Exception):
    """Base exception for EcoMCP"""
    
    def __init__(
        self,
        message: str,
        code: ErrorCode = ErrorCode.UNKNOWN_ERROR,
        details: Optional[Dict[str, Any]] = None
    ):
        self.message = message
        self.code = code
        self.details = details or {}
        super().__init__(self.message)
    
    def to_dict(self) -> Dict[str, Any]:
        """Convert to dictionary"""
        return {
            "error": self.message,
            "code": self.code.value,
            "details": self.details
        }


class ValidationError(EcoMCPException):
    """Validation error"""
    
    def __init__(self, message: str, details: Optional[Dict[str, Any]] = None):
        super().__init__(message, ErrorCode.VALIDATION_ERROR, details)


class InvalidInputError(EcoMCPException):
    """Invalid input error"""
    
    def __init__(self, message: str, details: Optional[Dict[str, Any]] = None):
        super().__init__(message, ErrorCode.INVALID_INPUT, details)


class APIError(EcoMCPException):
    """API error (e.g., OpenAI)"""
    
    def __init__(self, message: str, details: Optional[Dict[str, Any]] = None):
        super().__init__(message, ErrorCode.API_ERROR, details)


class NotFoundError(EcoMCPException):
    """Resource not found error"""
    
    def __init__(self, message: str, details: Optional[Dict[str, Any]] = None):
        super().__init__(message, ErrorCode.NOT_FOUND, details)


class ServerError(EcoMCPException):
    """Server error"""
    
    def __init__(self, message: str, details: Optional[Dict[str, Any]] = None):
        super().__init__(message, ErrorCode.SERVER_ERROR, details)


class TimeoutError(EcoMCPException):
    """Timeout error"""
    
    def __init__(self, message: str, details: Optional[Dict[str, Any]] = None):
        super().__init__(message, ErrorCode.TIMEOUT, details)


def format_error_response(
    error: Exception,
    code: ErrorCode = ErrorCode.UNKNOWN_ERROR
) -> Dict[str, Any]:
    """Format error as response dict"""
    if isinstance(error, EcoMCPException):
        return {
            "status": "error",
            **error.to_dict()
        }
    
    return {
        "status": "error",
        "error": str(error),
        "code": code.value,
        "details": {}
    }