""" 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)