Pathfinder MCP Server

"""Error handling utilities for Pathfinder MCP Server.""" from functools import wraps from typing import Any, Callable, TypeVar from pathfinder_mcp.logger import log_error, log_tool_call, log_tool_result T = TypeVar("T") class PathfinderError(Exception): """Base exception for Pathfinder errors.""" def __init__(self, message: str, code: str, **details: Any): super().__init__(message) self.message = message self.code = code self.details = details def to_dict(self) -> dict[str, Any]: """Convert to error response dict.""" result: dict[str, Any] = {"error": self.message, "code": self.code} if self.details: result.update(self.details) return result class SessionNotFoundError(PathfinderError): """Session not found.""" def __init__(self, session_id: str): super().__init__( message=f"Session not found: {session_id}", code="SESSION_NOT_FOUND", session_id=session_id, ) class InvalidPhaseError(PathfinderError): """Invalid phase transition or operation.""" def __init__(self, message: str, current_phase: str): super().__init__( message=message, code="INVALID_PHASE", current_phase=current_phase, ) class MissingArtifactError(PathfinderError): """Required artifact not found.""" def __init__(self, artifact: str, session_id: str): super().__init__( message=f"Missing artifact: {artifact}", code="MISSING_ARTIFACT", artifact=artifact, session_id=session_id, ) class ValidationError(PathfinderError): """Validation failed.""" def __init__(self, message: str, errors: list[str]): super().__init__( message=message, code="VALIDATION_ERROR", validation_errors=errors, ) def tool_error_handler( tool_name: str, ) -> Callable[[Callable[..., T]], Callable[..., T]]: """Decorator to wrap tools with error handling and logging. Args: tool_name: Name of the tool for logging Returns: Decorator function """ def decorator(func: Callable[..., T]) -> Callable[..., T]: @wraps(func) def sync_wrapper(*args: Any, **kwargs: Any) -> Any: session_id = kwargs.get("session_id") or (args[0] if args else None) log_tool_call(tool_name, session_id=session_id) try: result = func(*args, **kwargs) success = not (isinstance(result, dict) and "error" in result) log_tool_result( tool_name, success=success, session_id=session_id, error=result.get("error") if isinstance(result, dict) else None, ) return result except PathfinderError as e: log_tool_result( tool_name, success=False, session_id=session_id, error=e.message, ) return e.to_dict() except Exception as e: log_error(str(e), session_id=session_id, tool=tool_name) log_tool_result( tool_name, success=False, session_id=session_id, error=str(e), ) return {"error": str(e), "code": "INTERNAL_ERROR"} @wraps(func) async def async_wrapper(*args: Any, **kwargs: Any) -> Any: session_id = kwargs.get("session_id") or (args[0] if args else None) log_tool_call(tool_name, session_id=session_id) try: result = await func(*args, **kwargs) success = not (isinstance(result, dict) and "error" in result) log_tool_result( tool_name, success=success, session_id=session_id, error=result.get("error") if isinstance(result, dict) else None, ) return result except PathfinderError as e: log_tool_result( tool_name, success=False, session_id=session_id, error=e.message, ) return e.to_dict() except Exception as e: log_error(str(e), session_id=session_id, tool=tool_name) log_tool_result( tool_name, success=False, session_id=session_id, error=str(e), ) return {"error": str(e), "code": "INTERNAL_ERROR"} # Return appropriate wrapper based on function type import asyncio if asyncio.iscoroutinefunction(func): return async_wrapper # type: ignore return sync_wrapper # type: ignore return decorator