File size: 3,598 Bytes
f8ba6bf |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 |
"""
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)
|