File size: 2,145 Bytes
116524e
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
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
class ACEMCPError(Exception):
    """Base exception for ACE MCP extensions."""

    def __init__(self, message: str, code: str, details: dict | None = None):
        super().__init__(message)
        self.message = message
        self.code = code
        self.details = details or {}


class ValidationError(ACEMCPError):
    def __init__(self, message: str, details: dict | None = None):
        super().__init__(message, "ACE_MCP_VALIDATION_ERROR", details)


class SessionNotFoundError(ACEMCPError):
    def __init__(self, session_id: str):
        super().__init__(
            f"Session not found: {session_id}",
            "ACE_MCP_SESSION_NOT_FOUND",
            {"session_id": session_id},
        )


class ForbiddenInSafeModeError(ACEMCPError):
    def __init__(self, tool_name: str):
        super().__init__(
            f"Tool {tool_name} is forbidden in safe mode",
            "ACE_MCP_FORBIDDEN_IN_SAFE_MODE",
            {"tool_name": tool_name},
        )


class SaveLoadDisabledError(ACEMCPError):
    def __init__(self, tool_name: str):
        super().__init__(
            f"Tool {tool_name} is disabled (allow_save_load=false)",
            "ACE_MCP_SAVE_LOAD_DISABLED",
            {"tool_name": tool_name},
        )


class ProviderError(ACEMCPError):
    def __init__(self, message: str, details: dict | None = None):
        super().__init__(message, "ACE_MCP_PROVIDER_ERROR", details)


class TimeoutError(ACEMCPError):
    def __init__(self, message: str = "Operation timed out"):
        super().__init__(message, "ACE_MCP_TIMEOUT")


class InternalError(ACEMCPError):
    def __init__(self, message: str, details: dict | None = None):
        super().__init__(message, "ACE_MCP_INTERNAL_ERROR", details)


def map_error_to_mcp(err: Exception) -> dict:
    if isinstance(err, ACEMCPError):
        return {"code": err.code, "message": err.message, "details": err.details}
    return {
        "code": "ACE_MCP_INTERNAL_ERROR",
        "message": f"An unexpected error occurred: {str(err)}",
        "details": {"type": type(err).__name__},
    }