src/mcp_integration/exceptions.py

"""
DungeonMaster AI - MCP Integration Exceptions

Custom exception hierarchy for clear error handling in MCP operations.
"""

from __future__ import annotations


class MCPIntegrationError(Exception):
    """Base exception for all MCP integration errors."""

    def __init__(self, message: str = "An MCP integration error occurred") -> None:
        self.message = message
        super().__init__(self.message)


class MCPConnectionError(MCPIntegrationError):
    """Raised when connection to MCP server fails."""

    def __init__(
        self,
        message: str = "Failed to connect to MCP server",
        url: str | None = None,
    ) -> None:
        self.url = url
        if url:
            message = f"{message}: {url}"
        super().__init__(message)


class MCPTimeoutError(MCPIntegrationError):
    """Raised when an MCP operation times out."""

    def __init__(
        self,
        message: str = "MCP operation timed out",
        timeout_seconds: float | None = None,
        operation: str | None = None,
    ) -> None:
        self.timeout_seconds = timeout_seconds
        self.operation = operation
        if timeout_seconds and operation:
            message = f"{message}: {operation} after {timeout_seconds}s"
        elif timeout_seconds:
            message = f"{message} after {timeout_seconds}s"
        super().__init__(message)


class MCPToolNotFoundError(MCPIntegrationError):
    """Raised when a requested tool does not exist."""

    def __init__(self, tool_name: str) -> None:
        self.tool_name = tool_name
        super().__init__(f"Tool not found: {tool_name}")


class MCPToolExecutionError(MCPIntegrationError):
    """Raised when tool execution fails."""

    def __init__(
        self,
        tool_name: str,
        original_error: Exception | None = None,
        message: str | None = None,
    ) -> None:
        self.tool_name = tool_name
        self.original_error = original_error
        if message:
            error_msg = message
        elif original_error:
            error_msg = f"Tool '{tool_name}' execution failed: {original_error}"
        else:
            error_msg = f"Tool '{tool_name}' execution failed"
        super().__init__(error_msg)


class MCPUnavailableError(MCPIntegrationError):
    """Raised when the MCP server is unavailable."""

    def __init__(
        self,
        message: str = "MCP server is currently unavailable",
        reason: str | None = None,
    ) -> None:
        self.reason = reason
        if reason:
            message = f"{message}: {reason}"
        super().__init__(message)


class MCPCircuitBreakerOpenError(MCPIntegrationError):
    """Raised when the circuit breaker is open and rejecting requests."""

    def __init__(
        self,
        message: str = "Circuit breaker is open - too many recent failures",
        retry_after_seconds: float | None = None,
    ) -> None:
        self.retry_after_seconds = retry_after_seconds
        if retry_after_seconds:
            message = f"{message}. Retry after {retry_after_seconds:.1f}s"
        super().__init__(message)


class MCPInvalidResponseError(MCPIntegrationError):
    """Raised when MCP server returns an invalid or unexpected response."""

    def __init__(
        self,
        message: str = "Invalid response from MCP server",
        tool_name: str | None = None,
        response: object | None = None,
    ) -> None:
        self.tool_name = tool_name
        self.response = response
        if tool_name:
            message = f"{message} for tool '{tool_name}'"
        super().__init__(message)