Improve Error Messages #363 (#438)

* issue 363

Signed-off-by: Rakhi Dutta <[email protected]>

* issue 363

Signed-off-by: Rakhi Dutta <[email protected]>

* issue 363

Signed-off-by: Rakhi Dutta <[email protected]>

* flake8 error

Signed-off-by: Rakhi Dutta <[email protected]>

* flake8 error

Signed-off-by: Rakhi Dutta <[email protected]>

* flake8 error

Signed-off-by: Rakhi Dutta <[email protected]>

* makefile

Signed-off-by: Rakhi Dutta <[email protected]>

* issue 363

Signed-off-by: Rakhi Dutta <[email protected]>

* flake8

Signed-off-by: Rakhi Dutta <[email protected]>

* revert

Signed-off-by: Rakhi Dutta <[email protected]>

* revert

Signed-off-by: Rakhi Dutta <[email protected]>

---------

Signed-off-by: Rakhi Dutta <[email protected]>

Author: rakdutta

mcpgateway/admin.py

@@ -28,6 +28,8 @@
 from fastapi.responses import HTMLResponse, JSONResponse, RedirectResponse
 import httpx
 from sqlalchemy.orm import Session
+from sqlalchemy.exc import IntegrityError
+from pydantic import ValidationError
 
 # First-Party
 from mcpgateway.config import settings
@@ -67,6 +69,7 @@
 )
 from mcpgateway.utils.create_jwt_token import get_jwt_token
 from mcpgateway.utils.verify_credentials import require_auth, require_basic_auth
+from mcpgateway.utils.error_formatter import ErrorFormatter
 
 # Initialize services
 server_service = ServerService()
@@ -807,6 +810,7 @@ async def admin_add_gateway(request: Request, db: Session = Depends(get_db), use
         auth_header_key=form.get("auth_header_key", ""),
         auth_header_value=form.get("auth_header_value", ""),
     )
+
     try:
         await gateway_service.register_gateway(db, gateway)
         return JSONResponse(
@@ -821,6 +825,13 @@ async def admin_add_gateway(request: Request, db: Session = Depends(get_db), use
             return JSONResponse(content={"message": str(ex), "success": False}, status_code=400)
         if isinstance(ex, RuntimeError):
             return JSONResponse(content={"message": str(ex), "success": False}, status_code=500)
+        if isinstance(ex, ValidationError):
+            return JSONResponse(content=ErrorFormatter.format_validation_error(ex), status_code=422)
+        if isinstance(ex, IntegrityError):
+            return JSONResponse(
+                status_code=409,
+                content=ErrorFormatter.format_database_error(ex)
+            )
         return JSONResponse(content={"message": str(ex), "success": False}, status_code=500)
 
 

mcpgateway/main.py

@@ -53,6 +53,9 @@
 from sqlalchemy import text
 from sqlalchemy.orm import Session
 from starlette.middleware.base import BaseHTTPMiddleware
+from pydantic import ValidationError
+from sqlalchemy.exc import IntegrityError
+
 
 # First-Party
 from mcpgateway import __version__
@@ -128,6 +131,8 @@
     validate_request,
 )
 
+from mcpgateway.utils.error_formatter import ErrorFormatter
+
 # Import the admin routes from the new module
 from mcpgateway.version import router as version_router
 
@@ -244,6 +249,24 @@ async def lifespan(_app: FastAPI) -> AsyncIterator[None]:
 )
 
 
+# Global exceptions handlers
+
+@app.exception_handler(ValidationError)
+async def validation_exception_handler(request: Request, exc: ValidationError):
+    return JSONResponse(
+        status_code=422,
+        content=ErrorFormatter.format_validation_error(exc)
+    )
+
+
+@app.exception_handler(IntegrityError)
+async def database_exception_handler(request: Request, exc: IntegrityError):
+    return JSONResponse(
+        status_code=409,
+        content=ErrorFormatter.format_database_error(exc)
+    )
+
+
 class DocsAuthMiddleware(BaseHTTPMiddleware):
     """
     Middleware to protect FastAPI's auto-generated documentation routes
@@ -419,9 +442,7 @@ async def invalidate_resource_cache(uri: Optional[str] = None) -> None:
         resource_cache.clear()
 
 
-#################
 # Protocol APIs #
-#################
 @protocol_router.post("/initialize")
 async def initialize(request: Request, user: str = Depends(require_auth)) -> InitializeResult:
     """

mcpgateway/services/gateway_service.py

@@ -31,6 +31,7 @@
 from mcp.client.streamable_http import streamablehttp_client
 from sqlalchemy import select
 from sqlalchemy.orm import Session
+from sqlalchemy.exc import IntegrityError
 
 try:
     # Third-Party
@@ -187,6 +188,7 @@ async def register_gateway(self, db: Session, gateway: GatewayCreate) -> Gateway
             GatewayConnectionError: If there was an error connecting to the gateway
             ValueError: If required values are missing
             RuntimeError: If there is an error during processing that is not covered by other exceptions
+            IntegrityError: If there is a database integrity error
             BaseException: If an unexpected error occurs
 
         Examples:
@@ -281,6 +283,11 @@ async def register_gateway(self, db: Session, gateway: GatewayCreate) -> Gateway
                 re: ExceptionGroup[RuntimeError]
             logger.error(f"RuntimeErrors in group: {re.exceptions}")
             raise re.exceptions[0]
+        except* IntegrityError as ie:
+            if TYPE_CHECKING:
+                ie: ExceptionGroup[IntegrityError]
+            logger.error(f"IntegrityErrors in group: {ie.exceptions}")
+            raise ie.exceptions[0]
         except* BaseException as other:  # catches every other sub-exception
             if TYPE_CHECKING:
                 other: ExceptionGroup[BaseException]

mcpgateway/utils/error_formatter.py

@@ -0,0 +1,123 @@
+# -*- coding: utf-8 -*-
+"""MCP Gateway Centralized for Pydantic validation error, SQL exception.
+
+Copyright 2025
+SPDX-License-Identifier: Apache-2.0
+Authors: Mihai Criveti
+
+
+"""
+
+# Standard
+from typing import Dict, Any
+import logging
+
+# Third-Party
+from pydantic import ValidationError
+from sqlalchemy.exc import IntegrityError, DatabaseError
+
+
+logger = logging.getLogger(__name__)
+
+
+class ErrorFormatter:
+    """
+    Transform technical errors into user-friendly messages.
+    """
+    @staticmethod
+    def format_validation_error(error: ValidationError) -> Dict[str, Any]:
+        """
+        Convert Pydantic errors to user-friendly format.
+
+        Args:
+            error (ValidationError): The Pydantic validation error.
+
+        Returns:
+            Dict[str, Any]: A dictionary with formatted error details.
+        """
+        errors = []
+
+        for err in error.errors():
+            field = err.get('loc', ['field'])[-1]
+            msg = err.get('msg', 'Invalid value')
+
+            # Map technical messages to user-friendly ones
+            user_message = ErrorFormatter._get_user_message(field, msg)
+            errors.append({
+                "field": field,
+                "message": user_message
+            })
+
+        # Log the full error for debugging
+        logger.debug(f"Validation error: {error}")
+        print(type(error))
+
+        return {
+            "message": "Validation failed",
+            "details": errors,
+            "success": False
+        }
+
+    @staticmethod
+    def _get_user_message(field: str, technical_msg: str) -> str:
+        """
+        Map technical validation messages to user-friendly ones.
+
+        Args:
+            field (str): The field name.
+            technical_msg (str): The technical validation message.
+
+        Returns:
+            str: User-friendly error message.
+        """
+        mappings = {
+            "Tool name must start with a letter": f"{field.title()} must start with a letter and contain only letters, numbers, and underscores",
+            "Tool name exceeds maximum length": f"{field.title()} is too long (maximum 255 characters)",
+            "Tool URL must start with": f"{field.title()} must be a valid HTTP or WebSocket URL",
+            "cannot contain directory traversal": f"{field.title()} contains invalid characters",
+            "contains HTML tags": f"{field.title()} cannot contain HTML or script tags",
+        }
+
+        for pattern, friendly_msg in mappings.items():
+            if pattern in technical_msg:
+                return friendly_msg
+
+        # Default fallback
+        return f"Invalid {field}"
+
+    @staticmethod
+    def format_database_error(error: DatabaseError) -> Dict[str, Any]:
+        """
+        Convert database errors to user-friendly format.
+
+        Args:
+            error (DatabaseError): The database error.
+
+        Returns:
+            Dict[str, Any]: A dictionary with formatted error details.
+        """
+        error_str = str(error.orig) if hasattr(error, 'orig') else str(error)
+
+        # Log full error
+        logger.error(f"Database error: {error}")
+
+        # Map common database errors
+        if isinstance(error, IntegrityError):
+            if "UNIQUE constraint failed" in error_str:
+                if "gateways.url" in error_str:
+                    return {"message": "A gateway with this URL already exists", "success": False}
+                elif "gateways.name" in error_str:
+                    return {"message": "A gateway with this name already exists", "success": False}
+                elif "tools.name" in error_str:
+                    return {"message": "A tool with this name already exists", "success": False}
+                elif "resources.uri" in error_str:
+                    return {"message": "A resource with this URI already exists", "success": False}
+            elif "FOREIGN KEY constraint failed" in error_str:
+                return {"message": "Referenced item not found", "success": False}
+            elif "NOT NULL constraint failed" in error_str:
+                return {"message": "Required field is missing", "success": False}
+            elif "CHECK constraint failed:" in error_str:
+                return {"message": "Gateway validation failed. Please check the input data.", "success": False}
+
+        # Generic database error
+        return {"message": "Unable to complete the operation. Please try again.", "success": False}