syndarix/backend/app/services/mcp/exceptions.py

"""
MCP Exception Classes

Custom exceptions for MCP client operations with detailed error context.
"""

from typing import Any


class MCPError(Exception):
    """Base exception for all MCP-related errors."""

    def __init__(
        self,
        message: str,
        *,
        server_name: str | None = None,
        details: dict[str, Any] | None = None,
    ) -> None:
        super().__init__(message)
        self.message = message
        self.server_name = server_name
        self.details = details or {}

    def __str__(self) -> str:
        parts = [self.message]
        if self.server_name:
            parts.append(f"server={self.server_name}")
        if self.details:
            parts.append(f"details={self.details}")
        return " | ".join(parts)


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

    def __init__(
        self,
        message: str,
        *,
        server_name: str | None = None,
        url: str | None = None,
        cause: Exception | None = None,
        details: dict[str, Any] | None = None,
    ) -> None:
        super().__init__(message, server_name=server_name, details=details)
        self.url = url
        self.cause = cause

    def __str__(self) -> str:
        base = super().__str__()
        if self.url:
            base = f"{base} | url={self.url}"
        if self.cause:
            base = f"{base} | cause={type(self.cause).__name__}: {self.cause}"
        return base


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

    def __init__(
        self,
        message: str,
        *,
        server_name: str | None = None,
        timeout_seconds: float | None = None,
        operation: str | None = None,
        details: dict[str, Any] | None = None,
    ) -> None:
        super().__init__(message, server_name=server_name, details=details)
        self.timeout_seconds = timeout_seconds
        self.operation = operation

    def __str__(self) -> str:
        base = super().__str__()
        if self.timeout_seconds is not None:
            base = f"{base} | timeout={self.timeout_seconds}s"
        if self.operation:
            base = f"{base} | operation={self.operation}"
        return base


class MCPToolError(MCPError):
    """Raised when a tool execution fails."""

    def __init__(
        self,
        message: str,
        *,
        server_name: str | None = None,
        tool_name: str | None = None,
        tool_args: dict[str, Any] | None = None,
        error_code: str | None = None,
        details: dict[str, Any] | None = None,
    ) -> None:
        super().__init__(message, server_name=server_name, details=details)
        self.tool_name = tool_name
        self.tool_args = tool_args
        self.error_code = error_code

    def __str__(self) -> str:
        base = super().__str__()
        if self.tool_name:
            base = f"{base} | tool={self.tool_name}"
        if self.error_code:
            base = f"{base} | error_code={self.error_code}"
        return base


class MCPServerNotFoundError(MCPError):
    """Raised when a requested MCP server is not registered."""

    def __init__(
        self,
        server_name: str,
        *,
        available_servers: list[str] | None = None,
        details: dict[str, Any] | None = None,
    ) -> None:
        message = f"MCP server not found: {server_name}"
        super().__init__(message, server_name=server_name, details=details)
        self.available_servers = available_servers or []

    def __str__(self) -> str:
        base = super().__str__()
        if self.available_servers:
            base = f"{base} | available={self.available_servers}"
        return base


class MCPToolNotFoundError(MCPError):
    """Raised when a requested tool is not found on any server."""

    def __init__(
        self,
        tool_name: str,
        *,
        server_name: str | None = None,
        available_tools: list[str] | None = None,
        details: dict[str, Any] | None = None,
    ) -> None:
        message = f"Tool not found: {tool_name}"
        super().__init__(message, server_name=server_name, details=details)
        self.tool_name = tool_name
        self.available_tools = available_tools or []

    def __str__(self) -> str:
        base = super().__str__()
        if self.available_tools:
            base = f"{base} | available_tools={self.available_tools[:5]}..."
        return base


class MCPCircuitOpenError(MCPError):
    """Raised when a circuit breaker is open (server temporarily unavailable)."""

    def __init__(
        self,
        server_name: str,
        *,
        failure_count: int | None = None,
        reset_timeout: float | None = None,
        details: dict[str, Any] | None = None,
    ) -> None:
        message = f"Circuit breaker open for server: {server_name}"
        super().__init__(message, server_name=server_name, details=details)
        self.failure_count = failure_count
        self.reset_timeout = reset_timeout

    def __str__(self) -> str:
        base = super().__str__()
        if self.failure_count is not None:
            base = f"{base} | failures={self.failure_count}"
        if self.reset_timeout is not None:
            base = f"{base} | reset_in={self.reset_timeout}s"
        return base


class MCPValidationError(MCPError):
    """Raised when tool arguments fail validation."""

    def __init__(
        self,
        message: str,
        *,
        tool_name: str | None = None,
        field_errors: dict[str, str] | None = None,
        details: dict[str, Any] | None = None,
    ) -> None:
        super().__init__(message, details=details)
        self.tool_name = tool_name
        self.field_errors = field_errors or {}

    def __str__(self) -> str:
        base = super().__str__()
        if self.tool_name:
            base = f"{base} | tool={self.tool_name}"
        if self.field_errors:
            base = f"{base} | fields={list(self.field_errors.keys())}"
        return base