""" Domain exceptions for MCP Hangar. All domain-specific exceptions should be defined here. These exceptions carry context; the surface that answers decides the shape it is rendered in -- the REST error envelope in `server/api/middleware.py`, and the one MCP tool error payload in `application/mcp/tooling.py`. """ import math from typing import Any class MCPError(Exception): """Base exception for all MCP registry errors. Provides structured error information with context for debugging and logging. It carries the fields and serializes none of them. A `to_dict()` here used to render a fifth key, `type`, naming the class -- a second serializer that no caller reached and that therefore drifted out of agreement with the payload a tool answers with (#1509). A client reads one of two shapes, each built where it is answered: the REST envelope's `code`/`message`/`details`, or the tool error payload's `error`/`error_type`/`details` (`ToolErrorPayload`, #1495). """ def __init__( self, message: str, mcp_server_id: str = "", operation: str = "", details: dict[str, Any] | None = None, ): super().__init__(message) self.message = message self.mcp_server_id = mcp_server_id self.operation = operation self.details = details or {} def __repr__(self) -> str: return ( f"{self.__class__.__name__}(" f"message={self.message!r}, " f"mcp_server_id={self.mcp_server_id!r}, " f"operation={self.operation!r})" ) # --- McpServer Lifecycle Exceptions --- class McpServerError(MCPError): """Base exception for mcp_server-related errors.""" pass class McpServerNotFoundError(McpServerError): """Raised when a mcp_server is not found in the registry.""" def __init__(self, mcp_server_id: str): super().__init__( message=f"McpServer not found: {mcp_server_id}", mcp_server_id=mcp_server_id, operation="lookup", ) class McpServerStartError(McpServerError): """Raised when a mcp_server fails to start. Contains detailed diagnostics to help users understand and fix the issue: - reason: High-level reason for failure - stderr: Captured stderr output from the process (if available) - exit_code: Process exit code (if available) - suggestion: Actionable suggestion for fixing the issue """ def __init__( self, mcp_server_id: str, reason: str, details: dict[str, Any] | None = None, stderr: str | None = None, exit_code: int | None = None, suggestion: str | None = None, ): # Build user-friendly message message = f"Failed to start mcp_server: {reason}" if suggestion: message = f"{message}. Suggestion: {suggestion}" super().__init__( message=message, mcp_server_id=mcp_server_id, operation="start", details=details or {}, ) self.reason = reason self.stderr = stderr self.exit_code = exit_code self.suggestion = suggestion # Add diagnostics to details for structured logging/API responses if stderr: self.details["stderr"] = stderr[:2000] if len(stderr) > 2000 else stderr if exit_code is not None: self.details["exit_code"] = exit_code if suggestion: self.details["suggestion"] = suggestion def get_user_message(self) -> str: """Get a user-friendly error message with all available context.""" lines = [f"Failed to start mcp_server '{self.mcp_server_id}': {self.reason}"] if self.exit_code is not None: lines.append(f" Exit code: {self.exit_code}") if self.stderr: # Show first few lines of stderr stderr_lines = self.stderr.strip().split("\n")[:5] if stderr_lines: lines.append(" Process output:") for line in stderr_lines: lines.append(f" {line}") if len(self.stderr.strip().split("\n")) > 5: lines.append(" ... (truncated)") if self.suggestion: lines.append(f" Suggestion: {self.suggestion}") return "\n".join(lines) class CapabilityBlockedError(McpServerStartError): """A server with ``enforcement_mode`` block or quarantine serves a tool outside its ``expected_tools``. The start that finds it fails with this error, and the server goes DEAD for a capability block: no call starts it again and no group routes to it, and a deliberate start checks its tools again. A ``McpServerStartError``, so everything that handles a failed start handles this one the same way. The message names no tool. The names are the upstream's text, and the caller who is refused is not the operator who has to act on them. They are in the ``CapabilityViolationDetected`` event and the ``capability_drift_detected`` warning. """ def __init__(self, mcp_server_id: str, enforcement_mode: str = "block"): super().__init__( mcp_server_id=mcp_server_id, reason=( "capability_violation: it serves tools that are not in its declared expected_tools, " f"and its enforcement_mode is {enforcement_mode}" ), suggestion=( "fix the upstream or its capabilities.tools.expected_tools, then start the server " "explicitly; a start checks its tools again" ), ) self.enforcement_mode = enforcement_mode class McpServerDegradedError(McpServerError): """Raised when a mcp_server is in degraded state and cannot accept requests.""" def __init__( self, mcp_server_id: str, backoff_remaining: float = 0, consecutive_failures: int = 0, ): super().__init__( message=f"McpServer is degraded, retry in {backoff_remaining:.1f}s", mcp_server_id=mcp_server_id, operation="ensure_ready", details={ "backoff_remaining_s": backoff_remaining, "consecutive_failures": consecutive_failures, }, ) self.backoff_remaining = backoff_remaining self.consecutive_failures = consecutive_failures class CannotStartMcpServerError(McpServerError): """Raised when mcp_server cannot be started due to backoff or other constraints.""" def __init__(self, mcp_server_id: str, reason: str, time_until_retry: float = 0): super().__init__( message=f"Cannot start mcp_server: {reason}", mcp_server_id=mcp_server_id, operation="start", details={"time_until_retry_s": time_until_retry}, ) self.reason = reason self.time_until_retry = time_until_retry class McpServerNotHereError(McpServerError): """This server belongs to another instance of the fleet, not to this one. A `subprocess` or `docker` server is a child process of one gateway: its stdio is attached to that process and no peer has an address for it. A follower asked to start one refuses, and that refusal is **not a fault**. It used to travel as a generic start failure and reach the caller as a `500`, which says "this gateway is broken" about a gateway that is working exactly as designed -- measured on a two-replica deployment. A domain type rather than the launcher's own, so the model can let it through unwrapped and the API can answer `409`: the request was addressed to the wrong replica, which the caller can act on by asking another. """ def __init__(self, reason: str, mcp_server_id: str | None = None) -> None: super().__init__( message=reason, mcp_server_id=mcp_server_id or "", operation="start", ) class McpServerNotReadyError(McpServerError): """Raised when an operation requires READY state but mcp_server is not ready.""" def __init__(self, mcp_server_id: str, current_state: str): super().__init__( message=f"McpServer is not ready (state={current_state})", mcp_server_id=mcp_server_id, operation="invoke", details={"current_state": current_state}, ) self.current_state = current_state class InvalidStateTransitionError(McpServerError): """Raised when an invalid state transition is attempted.""" def __init__(self, mcp_server_id: str, from_state: str, to_state: str): super().__init__( message=f"Invalid state transition: {from_state} -> {to_state}", mcp_server_id=mcp_server_id, operation="transition", details={"from_state": from_state, "to_state": to_state}, ) self.from_state = from_state self.to_state = to_state # --- Tool Invocation Exceptions --- class ToolError(MCPError): """Base exception for tool-related errors.""" pass class ToolNotFoundError(ToolError): """Raised when a tool is not found in the mcp_server's catalog.""" def __init__(self, mcp_server_id: str, tool_name: str): super().__init__( message=f"Tool not found: {tool_name}", mcp_server_id=mcp_server_id, operation="invoke", details={"tool_name": tool_name}, ) self.tool_name = tool_name class ToolInvocationError(ToolError): """Raised when a tool invocation fails.""" def __init__(self, mcp_server_id: str, message: str, details: dict[str, Any] | None = None): super().__init__( message=message, mcp_server_id=mcp_server_id, operation="invoke", details=details or {}, ) class ToolCallFailedError(ToolInvocationError): """Raised by the facade's ``invoke`` for a call that did not succeed (#1453). ``invoke`` runs the call through the executor behind ``hangar_call``, so a call either refused by one of the configured controls or failed upstream arrives here. ``code`` is the ``error_type`` that ``hangar_call`` reports for the same call. A control's refusal codes include ``AuthorizationDenied``, ``ToolAccessDeniedError``, ``ToolWithdrawnError``, ``ToolDigestMismatchError``, ``ValidatorDenied``, ``TenantQuotaExceeded`` and ``CircuitBreakerOpen``. A failure's code is the name of the exception it raised. The message is the text ``hangar_call`` reports. """ def __init__(self, mcp_server_id: str, tool_name: str, code: str, message: str): super().__init__( mcp_server_id=mcp_server_id, message=message, details={"tool_name": tool_name, "code": code}, ) self.tool_name = tool_name self.code = code class ToolTimeoutError(ToolError): """Raised when a tool invocation times out.""" def __init__(self, mcp_server_id: str, tool_name: str, timeout: float): super().__init__( message=f"Tool invocation timed out after {timeout}s", mcp_server_id=mcp_server_id, operation="invoke", details={"tool_name": tool_name, "timeout_s": timeout}, ) self.tool_name = tool_name self.timeout = timeout class ToolAccessDeniedError(ToolError): """Raised when a tool is not accessible due to access policy. This is a config-driven denial, not an RBAC denial. The tool exists but is filtered out by the mcp_server's tool access policy. Note: Error message intentionally does not leak policy details. """ def __init__(self, mcp_server_id: str, tool_name: str): super().__init__( message="Tool not available for this mcp_server", mcp_server_id=mcp_server_id, operation="invoke", details={"tool_name": tool_name, "reason": "tool_not_in_access_policy"}, ) self.tool_name = tool_name class EgressPolicyDeniedError(ToolError): """Raised when an MCPEgressPolicy denies a tool call. Either the tool name matched a deny rule (or fell to a Deny default), or the arguments tripped a deterministic constraint (secret pattern / size limit). The caller-facing message is generic; the specific reason is carried in ``details`` for the audit trail. """ def __init__(self, mcp_server_id: str, tool_name: str, reason: str, policy_id: str | None = None): super().__init__( message="Tool call denied by egress policy", mcp_server_id=mcp_server_id, operation="invoke", details={"tool_name": tool_name, "reason": reason, "policy_id": policy_id}, ) self.tool_name = tool_name self.reason = reason #: Content hash of the policy that produced this verdict (#1129), so an #: audit record says which policy denied and not only that one did. self.policy_id = policy_id class EgressPolicyApprovalRequiredError(ToolError): """Raised when an MCPEgressPolicy routes a tool call to approval. On the governed invoke path (#921) the approval gate asks a human first and only a granted, revalidated approval converts the verdict; this is raised when the gate is not configured, refused, or timed out -- the fail-closed default. """ def __init__(self, mcp_server_id: str, tool_name: str, policy_id: str | None = None): super().__init__( message="Tool call requires approval", mcp_server_id=mcp_server_id, operation="invoke", details={"tool_name": tool_name, "reason": "require_approval", "policy_id": policy_id}, ) self.tool_name = tool_name #: Content hash of the policy that routed this call to approval (#1129). self.policy_id = policy_id # --- Client/Communication Exceptions --- class ClientError(MCPError): """Raised when the stdio client encounters an error.""" def __init__( self, message: str, mcp_server_id: str = "", details: dict[str, Any] | None = None, ): super().__init__( message=message, mcp_server_id=mcp_server_id, operation="client", details=details or {}, ) class ClientNotConnectedError(ClientError): """Raised when attempting to use a client that is not connected.""" def __init__(self, mcp_server_id: str = ""): super().__init__(message="Client is not connected", mcp_server_id=mcp_server_id) class ClientTimeoutError(ClientError): """Raised when a client operation times out.""" def __init__(self, mcp_server_id: str = "", timeout: float = 0, operation: str = "call"): super().__init__( message=f"Client operation timed out after {timeout}s", mcp_server_id=mcp_server_id, details={"timeout_s": timeout, "operation": operation}, ) self.timeout = timeout # --- Validation Exceptions --- class ValidationError(MCPError): """Raised when input validation fails.""" def __init__( self, message: str, field: str = "", value: Any = None, details: dict[str, Any] | None = None, ): base_details = {"field": field} if value is not None: # Sanitize value for logging (truncate if too long) str_value = str(value) if len(str_value) > 100: str_value = str_value[:100] + "..." base_details["value"] = str_value if details: base_details.update(details) super().__init__(message=message, operation="validation", details=base_details) self.field = field self.value = value class ConfigurationError(MCPError): """Raised when configuration is invalid. This is an operator-input problem (a typo in the config, an unparseable block): the API answers it 500, because it is not a request the caller can retry until a human fixes the file. See ``ConfigurationUnavailableError`` for the narrower, retryable I/O case that maps to 503. """ def __init__(self, message: str, details: dict[str, Any] | None = None): super().__init__(message=message, operation="configuration", details=details or {}) class ConfigurationRestartRequiredError(ConfigurationError): """A configuration the running process cannot apply, and a restart can. A reload refuses a file that changes ``tool_access.mode``: the front-door tool surface is built at boot, so a live change would leave the surface and the access rules disagreeing (#1424). The reload changes nothing, and the API answers 409 -- the file is not wrong and nothing broke, it is only something this process cannot take without a restart. """ class ConfigurationUnavailableError(ConfigurationError): """A configuration operation failed on a transient/unavailable I/O condition. Distinct from a plain ``ConfigurationError`` (an operator-input problem, a 500): this names a genuine "the filesystem said no" case -- the rotating backup cannot be written because the configuration file's directory is not writable by the gateway process. That is retryable once the environment is fixed, so the API answers it 503 with a message saying what to fix, rather than a 500 that reads as "the gateway is broken". Only this narrow subclass maps to 503; a generic ``ConfigurationError`` (e.g. the reload fault-barrier or a bad capabilities block) does not, so an operator-input error is not dressed up as a retryable outage. """ # --- Rate Limiting Exceptions --- #: Whose budget a `RateLimitExceeded` refusal found used up (#1471): the #: caller's own, or the one every caller shares. RATE_LIMIT_CALLER = "caller" RATE_LIMIT_ALL_CALLERS = "all_callers" _RATE_LIMIT_BUDGETS = { RATE_LIMIT_CALLER: "this caller's rate limit", RATE_LIMIT_ALL_CALLERS: "the rate limit all callers share", } class RateLimitExceeded(MCPError): """Raised when rate limit is exceeded. A refusal given a `retry_after` reads the same on every path (#1471). Its message, which is all an MCP tool result carries, names the code, the budget, the limit and when to retry, and `details` holds the same values for the HTTP API. """ def __init__( self, mcp_server_id: str = "", limit: int = 0, window_seconds: int = 0, *, retry_after: float | None = None, scope: str = "", key: str = "", rps: float | None = None, ): details: dict[str, Any] = {"limit": limit, "window_seconds": window_seconds} if retry_after is None: message = f"Rate limit exceeded: {limit} requests per {window_seconds}s" else: retry_after = max(0.0, retry_after) window_seconds = window_seconds or math.ceil(retry_after) budget = _RATE_LIMIT_BUDGETS.get(scope, "the rate limit") refill = f", refilled at {rps:g} per second" if rps is not None else "" message = ( f"RateLimitExceeded: {budget} for {key or 'this call'} is used up " f"({limit} at once{refill}). Retry after {retry_after:.2f}s." ) details = { "limit": limit, "window_seconds": window_seconds, "retry_after": round(retry_after, 3), "scope": scope, "key": key, "rps": rps, } super().__init__( message=message, mcp_server_id=mcp_server_id, operation="rate_limit", details=details, ) self.limit = limit self.window_seconds = window_seconds self.retry_after = retry_after self.scope = scope # --- Authentication Exceptions --- class AuthenticationError(MCPError): """Base class for authentication errors. All authentication-related failures inherit from this class, enabling unified handling of auth errors. """ def __init__( self, message: str, auth_method: str = "", details: dict[str, Any] | None = None, ): super().__init__( message=message, operation="authentication", details={"auth_method": auth_method, **(details or {})}, ) self.auth_method = auth_method class InvalidCredentialsError(AuthenticationError): """Credentials are invalid or malformed. Raised when: - API key format is invalid - JWT signature verification fails - Token is malformed - Unknown API key """ def __init__( self, message: str = "Invalid credentials", auth_method: str = "", details: dict[str, Any] | None = None, ): super().__init__( message=message, auth_method=auth_method, details=details, ) class ExpiredCredentialsError(AuthenticationError): """Credentials have expired. Raised when: - JWT exp claim is in the past - API key has passed its expiration date """ def __init__( self, message: str = "Credentials have expired", auth_method: str = "", expired_at: float | None = None, ): super().__init__( message=message, auth_method=auth_method, details={"expired_at": expired_at} if expired_at else None, ) self.expired_at = expired_at class RevokedCredentialsError(AuthenticationError): """Credentials have been revoked. Raised when: - API key has been explicitly revoked - JWT is on a revocation list """ def __init__( self, message: str = "Credentials have been revoked", auth_method: str = "", revoked_at: float | None = None, ): super().__init__( message=message, auth_method=auth_method, details={"revoked_at": revoked_at} if revoked_at else None, ) self.revoked_at = revoked_at class MissingCredentialsError(AuthenticationError): """No credentials provided when authentication is required. Raised when: - No Authorization header present - No API key header present - Authentication is required but allow_anonymous is False """ def __init__( self, message: str = "No credentials provided", expected_methods: list[str] | None = None, ): super().__init__( message=message, auth_method="none", details={"expected_methods": expected_methods} if expected_methods else None, ) self.expected_methods = expected_methods or [] class RateLimitExceededError(AuthenticationError): """Rate limit exceeded for authentication attempts. Raised when: - Too many failed authentication attempts from an IP - IP is temporarily locked out """ def __init__( self, message: str = "Rate limit exceeded", retry_after: float | None = None, ): super().__init__( message=message, auth_method="rate_limit", details={"retry_after": retry_after} if retry_after else None, ) self.retry_after = retry_after class TokenLifetimeExceededError(AuthenticationError): """JWT token lifetime exceeds the configured maximum. Raised when: - Token lifetime (exp - iat) exceeds max_token_lifetime - Prevents excessively long-lived tokens from being accepted """ def __init__( self, actual_lifetime: float, max_lifetime: float, ): message = f"JWT token lifetime exceeds maximum allowed: {actual_lifetime}s > {max_lifetime}s" super().__init__( message=message, auth_method="jwt", details={ "actual_lifetime": actual_lifetime, "max_lifetime": max_lifetime, }, ) self.actual_lifetime = actual_lifetime self.max_lifetime = max_lifetime # --- Authorization Exceptions --- class AuthorizationError(MCPError): """Base class for authorization errors. All authorization-related failures inherit from this class. """ def __init__( self, message: str, principal_id: str = "", action: str = "", resource: str = "", details: dict[str, Any] | None = None, ): super().__init__( message=message, operation="authorization", details={ "principal_id": principal_id, "action": action, "resource": resource, **(details or {}), }, ) self.principal_id = principal_id self.action = action self.resource = resource class AccessDeniedError(AuthorizationError): """Principal does not have permission for the requested action. The most common authorization error - principal is authenticated but lacks the necessary permissions. """ def __init__( self, principal_id: str, action: str, resource: str, reason: str = "", ): message = f"Access denied: {principal_id} cannot {action} on {resource}" if reason: message = f"{message} ({reason})" super().__init__( message=message, principal_id=principal_id, action=action, resource=resource, details={"reason": reason} if reason else None, ) self.reason = reason class RoleNotFoundError(AuthorizationError): """Raised when a role is not found in the store.""" def __init__(self, role_name: str): super().__init__( message=f"Role not found: {role_name}", details={"role_name": role_name}, ) self.role_name = role_name class CannotModifyBuiltinRoleError(AuthorizationError): """Raised when an attempt is made to modify or delete a built-in role.""" def __init__(self, role_name: str): super().__init__( message=f"Cannot modify built-in role: {role_name}", details={"role_name": role_name}, ) self.role_name = role_name # --- Multi-Tenancy Exceptions --- # --- Event Store Exceptions --- class CompactionError(MCPError): """Raised when event stream compaction fails. Compaction requires a snapshot to exist for the stream. If no snapshot is found, this error is raised to prevent data loss (compaction without a snapshot would destroy all events with no way to rebuild aggregate state). """ def __init__(self, stream_id: str, reason: str): super().__init__( message=f"Cannot compact stream '{stream_id}': {reason}", operation="compact_stream", details={"stream_id": stream_id, "reason": reason}, ) self.stream_id = stream_id self.reason = reason # --- Registry Exceptions --- class RegistryError(MCPError): """Base exception for registry-related errors.""" def __init__( self, message: str, details: dict[str, Any] | None = None, ): super().__init__( message=message, operation="registry", details=details or {}, ) class RegistryConnectionError(RegistryError): """Failed to connect to the registry.""" def __init__(self, url: str, reason: str): super().__init__( message=f"Failed to connect to registry: {reason}", details={"url": url, "reason": reason}, ) self.url = url self.reason = reason class RegistryServerNotFoundError(RegistryError): """Server not found in the registry.""" def __init__(self, server_id: str): super().__init__( message=f"Server not found in registry: {server_id}", details={"server_id": server_id}, ) self.server_id = server_id class RegistryAmbiguousSearchError(RegistryError): """Multiple servers match the search query.""" def __init__(self, query: str, matches: list[str]): super().__init__( message=f"Ambiguous search '{query}': found {len(matches)} matches", details={"query": query, "matches": matches}, ) self.query = query self.matches = matches # --- Installation Exceptions --- class InstallationError(MCPError): """Base exception for package installation errors.""" def __init__( self, message: str, package: str = "", details: dict[str, Any] | None = None, ): super().__init__( message=message, operation="installation", details={"package": package, **(details or {})}, ) self.package = package class MissingSecretsError(MCPError): """Required secrets are not available.""" def __init__(self, mcp_server_name: str, missing: list[str], instructions: str | None = None): super().__init__( message=f"Missing required secrets for '{mcp_server_name}': {', '.join(missing)}", operation="secrets", details={ "mcp_server_name": mcp_server_name, "missing": missing, "instructions": instructions, }, ) self.mcp_server_name = mcp_server_name self.missing = missing self.instructions = instructions class UnverifiedMcpServerError(MCPError): """Attempted to load an unverified mcp_server without explicit flag.""" def __init__(self, mcp_server_name: str): super().__init__( message=f"McpServer '{mcp_server_name}' is not verified. Use force_unverified=True to load.", operation="load", details={"mcp_server_name": mcp_server_name}, ) self.mcp_server_name = mcp_server_name class McpServerAlreadyLoadedError(MCPError): """McpServer is already loaded.""" def __init__(self, mcp_server_id: str): super().__init__( message=f"McpServer '{mcp_server_id}' is already loaded", mcp_server_id=mcp_server_id, operation="load", ) class McpServerNotHotLoadedError(MCPError): """Cannot unload a mcp_server that was not hot-loaded.""" def __init__(self, mcp_server_id: str): super().__init__( message=f"McpServer '{mcp_server_id}' was not hot-loaded and cannot be unloaded", mcp_server_id=mcp_server_id, operation="unload", ) # legacy aliases globals().update( { "".join(("Pro", "viderError")): McpServerError, "".join(("Pro", "viderNotFoundError")): McpServerNotFoundError, "".join(("Pro", "viderStartError")): McpServerStartError, "".join(("Pro", "viderDegradedError")): McpServerDegradedError, "".join(("CannotStartPro", "viderError")): CannotStartMcpServerError, "".join(("Pro", "viderNotReadyError")): McpServerNotReadyError, "".join(("UnverifiedPro", "viderError")): UnverifiedMcpServerError, "".join(("Pro", "viderAlreadyLoadedError")): McpServerAlreadyLoadedError, "".join(("Pro", "viderNotHotLoadedError")): McpServerNotHotLoadedError, } )