error_handler.py•4.61 kB
"""
Error handling utilities for the MCP server
"""
import logging
import functools
from typing import Any, Callable, Dict, List
from mcp.types import TextContent
logger = logging.getLogger(__name__)
class MCPError(Exception):
"""Base exception for MCP server errors"""
def __init__(self, message: str, error_code: str = "GENERAL_ERROR", details: Dict[str, Any] = None):
self.message = message
self.error_code = error_code
self.details = details or {}
super().__init__(self.message)
class JiraError(MCPError):
"""Jira-specific errors"""
def __init__(self, message: str, status_code: int = None, details: Dict[str, Any] = None):
super().__init__(message, "JIRA_ERROR", details)
self.status_code = status_code
class GitLabError(MCPError):
"""GitLab-specific errors"""
def __init__(self, message: str, status_code: int = None, details: Dict[str, Any] = None):
super().__init__(message, "GITLAB_ERROR", details)
self.status_code = status_code
class ConfigurationError(MCPError):
"""Configuration-related errors"""
def __init__(self, message: str, details: Dict[str, Any] = None):
super().__init__(message, "CONFIG_ERROR", details)
class AuthenticationError(MCPError):
"""Authentication-related errors"""
def __init__(self, message: str, service: str = None, details: Dict[str, Any] = None):
super().__init__(message, "AUTH_ERROR", details)
self.service = service
def handle_errors(func: Callable) -> Callable:
"""
Decorator to handle errors in MCP tool functions
Converts exceptions to proper MCP error responses
"""
@functools.wraps(func)
async def wrapper(*args, **kwargs) -> List[TextContent]:
try:
return await func(*args, **kwargs)
except MCPError as e:
logger.error(f"MCP Error in {func.__name__}: {e.message}")
return [TextContent(
type="text",
text=f"Error: {e.message}\nCode: {e.error_code}\nDetails: {e.details}"
)]
except Exception as e:
logger.error(f"Unexpected error in {func.__name__}: {str(e)}", exc_info=True)
return [TextContent(
type="text",
text=f"Unexpected error: {str(e)}"
)]
return wrapper
def retry_on_failure(max_retries: int = 3, delay: float = 1.0, backoff_factor: float = 2.0):
"""
Decorator to retry failed operations with exponential backoff
"""
import asyncio
def decorator(func: Callable) -> Callable:
@functools.wraps(func)
async def wrapper(*args, **kwargs):
last_exception = None
current_delay = delay
for attempt in range(max_retries + 1):
try:
return await func(*args, **kwargs)
except Exception as e:
last_exception = e
if attempt == max_retries:
logger.error(f"Function {func.__name__} failed after {max_retries} retries: {str(e)}")
raise
logger.warning(f"Attempt {attempt + 1} failed for {func.__name__}: {str(e)}. Retrying in {current_delay}s...")
await asyncio.sleep(current_delay)
current_delay *= backoff_factor
raise last_exception
return wrapper
return decorator
def validate_required_fields(data: Dict[str, Any], required_fields: List[str]) -> None:
"""
Validate that required fields are present in data
"""
missing_fields = [field for field in required_fields if field not in data or data[field] is None]
if missing_fields:
raise MCPError(
f"Missing required fields: {', '.join(missing_fields)}",
"VALIDATION_ERROR",
{"missing_fields": missing_fields}
)
def safe_get_nested(data: Dict[str, Any], path: str, default: Any = None) -> Any:
"""
Safely get nested dictionary values using dot notation
Example: safe_get_nested(data, "fields.assignee.displayName", "Unassigned")
"""
try:
keys = path.split('.')
value = data
for key in keys:
if isinstance(value, dict) and key in value:
value = value[key]
else:
return default
return value if value is not None else default
except Exception:
return default