Clean up unused error classes

Author: praboud-ant

src/mcp/server/auth/errors.py

@@ -4,9 +4,15 @@
 Corresponds to TypeScript file: src/server/auth/errors.ts
 """
 
-from typing import Dict
+from typing import Literal
 
-from pydantic import ValidationError
+from pydantic import BaseModel, ValidationError
+
+ErrorCode = Literal["invalid_request", "invalid_client"]
+
+class ErrorResponse(BaseModel):
+    error: ErrorCode
+    error_description: str
 
 
 class OAuthError(Exception):
@@ -16,25 +22,17 @@ class OAuthError(Exception):
     Corresponds to OAuthError in src/server/auth/errors.ts
     """
 
-    error_code: str = "server_error"
-
-    def __init__(self, message: str):
-        super().__init__(message)
-        self.message = message
-
-    def to_response_object(self) -> Dict[str, str]:
-        """Convert error to JSON response object."""
-        return {"error": self.error_code, "error_description": self.message}
+    error_code: ErrorCode
 
+    def __init__(self, error_description: str):
+        super().__init__(error_description)
+        self.error_description = error_description
 
-class ServerError(OAuthError):
-    """
-    Server error.
-
-    Corresponds to ServerError in src/server/auth/errors.ts
-    """
-
-    error_code = "server_error"
+    def error_response(self) -> ErrorResponse:
+        return ErrorResponse(
+            error=self.error_code,
+            error_description=self.error_description,
+        )
 
 
 class InvalidRequestError(OAuthError):
@@ -57,96 +55,6 @@ class InvalidClientError(OAuthError):
     error_code = "invalid_client"
 
 
-class InvalidGrantError(OAuthError):
-    """
-    Invalid grant error.
-
-    Corresponds to InvalidGrantError in src/server/auth/errors.ts
-    """
-
-    error_code = "invalid_grant"
-
-
-class UnauthorizedClientError(OAuthError):
-    """
-    Unauthorized client error.
-
-    Corresponds to UnauthorizedClientError in src/server/auth/errors.ts
-    """
-
-    error_code = "unauthorized_client"
-
-
-class UnsupportedGrantTypeError(OAuthError):
-    """
-    Unsupported grant type error.
-
-    Corresponds to UnsupportedGrantTypeError in src/server/auth/errors.ts
-    """
-
-    error_code = "unsupported_grant_type"
-
-
-class UnsupportedResponseTypeError(OAuthError):
-    """
-    Unsupported response type error.
-
-    Corresponds to UnsupportedResponseTypeError in src/server/auth/errors.ts
-    """
-
-    error_code = "unsupported_response_type"
-
-
-class InvalidScopeError(OAuthError):
-    """
-    Invalid scope error.
-
-    Corresponds to InvalidScopeError in src/server/auth/errors.ts
-    """
-
-    error_code = "invalid_scope"
-
-
-class AccessDeniedError(OAuthError):
-    """
-    Access denied error.
-
-    Corresponds to AccessDeniedError in src/server/auth/errors.ts
-    """
-
-    error_code = "access_denied"
-
-
-class TemporarilyUnavailableError(OAuthError):
-    """
-    Temporarily unavailable error.
-
-    Corresponds to TemporarilyUnavailableError in src/server/auth/errors.ts
-    """
-
-    error_code = "temporarily_unavailable"
-
-
-class InvalidTokenError(OAuthError):
-    """
-    Invalid token error.
-
-    Corresponds to InvalidTokenError in src/server/auth/errors.ts
-    """
-
-    error_code = "invalid_token"
-
-
-class InsufficientScopeError(OAuthError):
-    """
-    Insufficient scope error.
-
-    Corresponds to InsufficientScopeError in src/server/auth/errors.ts
-    """
-
-    error_code = "insufficient_scope"
-
-
 def stringify_pydantic_error(validation_error: ValidationError) -> str:
     return "\n".join(
         f"{'.'.join(str(loc) for loc in e['loc'])}: {e['msg']}"

src/mcp/server/auth/handlers/authorize.py

@@ -216,7 +216,7 @@ async def error_response(
                 # For redirect_uri validation errors, return direct error (no redirect)
                 return await error_response(
                     error="invalid_request",
-                    error_description=validation_error.message,
+                    error_description=validation_error.error_description,
                 )
 
             # Validate scope - for scope errors, we can redirect
@@ -226,7 +226,7 @@ async def error_response(
                 # For scope errors, redirect with error parameters
                 return await error_response(
                     error="invalid_scope",
-                    error_description=validation_error.message,
+                    error_description=validation_error.error_description,
                 )
 
             # Setup authorization parameters

src/mcp/server/auth/handlers/register.py

@@ -83,9 +83,9 @@ async def registration_handler(request: Request) -> Response:
             software_version=client_metadata.software_version,
         )
         # Register client
-        client = await clients_store.register_client(client_info)
+        await clients_store.register_client(client_info)
 
         # Return client information
-        return PydanticJSONResponse(content=client, status_code=201)
+        return PydanticJSONResponse(content=client_info, status_code=201)
 
     return registration_handler

src/mcp/server/auth/handlers/revoke.py

@@ -11,6 +11,7 @@
 from starlette.responses import Response
 
 from mcp.server.auth.errors import (
+    InvalidClientError,
     stringify_pydantic_error,
 )
 from mcp.server.auth.json_response import PydanticJSONResponse
@@ -46,7 +47,10 @@ async def revocation_handler(request: Request) -> Response:
             )
 
         # Authenticate client
-        client_auth_result = await client_authenticator(revocation_request)
+        try:
+            client_auth_result = await client_authenticator(revocation_request)
+        except InvalidClientError as e:
+            return PydanticJSONResponse(status_code=401, content=e.error_response())
 
         # Revoke token
         if provider.revoke_token:

src/mcp/server/auth/handlers/token.py

@@ -13,6 +13,8 @@
 from starlette.requests import Request
 
 from mcp.server.auth.errors import (
+    ErrorResponse,
+    InvalidClientError,
     stringify_pydantic_error,
 )
 from mcp.server.auth.json_response import PydanticJSONResponse
@@ -53,7 +55,7 @@ class TokenRequest(RootModel):
 def create_token_handler(
     provider: OAuthServerProvider, client_authenticator: ClientAuthenticator
 ) -> Callable:
-    def response(obj: TokenSuccessResponse | TokenErrorResponse):
+    def response(obj: TokenSuccessResponse | TokenErrorResponse | ErrorResponse):
         status_code = 200
         if isinstance(obj, TokenErrorResponse):
             status_code = 400
@@ -78,7 +80,11 @@ async def token_handler(request: Request):
                     error_description=stringify_pydantic_error(validation_error),
                 )
             )
-        client_info = await client_authenticator(token_request)
+        
+        try:
+            client_info = await client_authenticator(token_request)
+        except InvalidClientError as e:
+            return response(e.error_response())
 
         if token_request.grant_type not in client_info.grant_types:
             return response(
@@ -124,8 +130,9 @@ async def token_handler(request: Request):
                         TokenErrorResponse(
                             error="invalid_request",
                             error_description=(
-                        "redirect_uri didn't match the one used when creating auth code"
-                    ),
+                                "redirect_uri did not match the one "
+                                "used when creating auth code"
+                            ),
                         )
                     )
 

src/mcp/server/auth/middleware/bearer_auth.py

@@ -16,7 +16,6 @@
 from starlette.requests import HTTPConnection
 from starlette.types import Scope
 
-from mcp.server.auth.errors import InsufficientScopeError, InvalidTokenError, OAuthError
 from mcp.server.auth.provider import OAuthServerProvider
 from mcp.server.auth.types import AuthInfo
 
@@ -51,21 +50,17 @@ async def authenticate(self, conn: HTTPConnection):
 
         token = auth_header[7:]  # Remove "Bearer " prefix
 
-        try:
-            # Validate the token with the provider
-            auth_info = await self.provider.load_access_token(token)
+        # Validate the token with the provider
+        auth_info = await self.provider.load_access_token(token)
 
-            if not auth_info:
-                raise InvalidTokenError("Invalid access token")
+        if not auth_info:
+            return None
 
-            if auth_info.expires_at and auth_info.expires_at < int(time.time()):
-                raise InvalidTokenError("Token has expired")
+        if auth_info.expires_at and auth_info.expires_at < int(time.time()):
+            return None
 
-            return AuthCredentials(auth_info.scopes), AuthenticatedUser(auth_info)
+        return AuthCredentials(auth_info.scopes), AuthenticatedUser(auth_info)
 
-        except (InvalidTokenError, InsufficientScopeError, OAuthError):
-            # Return None to indicate authentication failure
-            return None
 
 
 class RequireAuthMiddleware:

src/mcp/server/auth/middleware/client_auth.py

@@ -5,11 +5,9 @@
 """
 
 import time
-from typing import Any, Callable, Dict, Optional
+from typing import Optional
 
 from pydantic import BaseModel
-from starlette.exceptions import HTTPException
-from starlette.requests import Request
 
 from mcp.server.auth.errors import (
     InvalidClientError,

tests/server/fastmcp/auth/test_auth_integration.py

@@ -18,7 +18,6 @@
 from starlette.applications import Starlette
 from starlette.routing import Mount
 
-from mcp.server.auth.errors import InvalidTokenError
 from mcp.server.auth.provider import (
     AuthorizationCode,
     AuthorizationParams,
@@ -97,8 +96,7 @@ async def load_authorization_code(
     async def exchange_authorization_code(
         self, client: OAuthClientInformationFull, authorization_code: AuthorizationCode
     ) -> TokenSuccessResponse:
-        if authorization_code.code not in self.auth_codes:
-            raise InvalidTokenError("Invalid authorization code")
+        assert authorization_code.code in self.auth_codes
 
         # Generate an access token and refresh token
         access_token = f"access_{secrets.token_hex(32)}"
@@ -152,19 +150,16 @@ async def exchange_refresh_token(
         scopes: List[str],
     ) -> TokenSuccessResponse:
         # Check if refresh token exists
-        if refresh_token.token not in self.refresh_tokens:
-            raise InvalidTokenError("Invalid refresh token")
+        assert refresh_token.token in self.refresh_tokens
 
         old_access_token = self.refresh_tokens[refresh_token.token]
 
         # Check if the access token exists
-        if old_access_token not in self.tokens:
-            raise InvalidTokenError("Invalid refresh token")
+        assert old_access_token in self.tokens
 
         # Check if the token was issued to this client
         token_info = self.tokens[old_access_token]
-        if token_info.client_id != client.client_id:
-            raise InvalidTokenError("Refresh token was not issued to this client")
+        assert token_info.client_id == client.client_id
 
         # Generate a new access token and refresh token
         new_access_token = f"access_{secrets.token_hex(32)}"
@@ -1017,6 +1012,18 @@ def test_tool(x: int) -> str:
         # TODO: we should return 401/403 depending on whether authn or authz fails
         assert response.status_code == 403, response.content
 
+        response = await test_client.post(
+            "/messages/",
+            headers={"Authorization": "invalid"},
+        )
+        assert response.status_code == 403
+
+        response = await test_client.post(
+            "/messages/",
+            headers={"Authorization": "Bearer invalid"},
+        )
+        assert response.status_code == 403
+
         # now, become authenticated and try to go through the flow again
         client_metadata = {
             "redirect_uris": ["https://client.example.com/callback"],