Move around validation logic

Author: praboud-ant

src/mcp/server/auth/errors.py

@@ -28,22 +28,6 @@ def error_response(self) -> ErrorResponse:
         )
 
 
-class InvalidRequestError(OAuthError):
-    """
-    Invalid request error.
-    """
-
-    error_code = "invalid_request"
-
-
-class InvalidClientError(OAuthError):
-    """
-    Invalid client error.
-    """
-
-    error_code = "invalid_client"
-
-
 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

@@ -9,7 +9,6 @@
 from starlette.responses import RedirectResponse, Response
 
 from mcp.server.auth.errors import (
-    InvalidRequestError,
     OAuthError,
     stringify_pydantic_error,
 )
@@ -19,7 +18,10 @@
     OAuthServerProvider,
     construct_redirect_uri,
 )
-from mcp.shared.auth import OAuthClientInformationFull
+from mcp.shared.auth import (
+    InvalidRedirectUriError,
+    InvalidScopeError,
+)
 
 logger = logging.getLogger(__name__)
 
@@ -66,37 +68,6 @@ class AuthorizationErrorResponse(BaseModel):
     state: str | None = None
 
 
-def validate_scope(
-    requested_scope: str | None, client: OAuthClientInformationFull
-) -> list[str] | None:
-    if requested_scope is None:
-        return None
-    requested_scopes = requested_scope.split(" ")
-    allowed_scopes = [] if client.scope is None else client.scope.split(" ")
-    for scope in requested_scopes:
-        if scope not in allowed_scopes:
-            raise InvalidRequestError(f"Client was not registered with scope {scope}")
-    return requested_scopes
-
-
-def validate_redirect_uri(
-    redirect_uri: AnyHttpUrl | None, client: OAuthClientInformationFull
-) -> AnyHttpUrl:
-    if redirect_uri is not None:
-        # Validate redirect_uri against client's registered redirect URIs
-        if redirect_uri not in client.redirect_uris:
-            raise InvalidRequestError(
-                f"Redirect URI '{redirect_uri}' not registered for client"
-            )
-        return redirect_uri
-    elif len(client.redirect_uris) == 1:
-        return client.redirect_uris[0]
-    else:
-        raise InvalidRequestError(
-            "redirect_uri must be specified when client has multiple registered URIs"
-        )
-
-
 def best_effort_extract_string(
     key: str, params: None | FormData | QueryParams
 ) -> str | None:
@@ -146,8 +117,8 @@ async def error_response(
                         best_effort_extract_string("redirect_uri", params)
                     ).root
                 try:
-                    redirect_uri = validate_redirect_uri(raw_redirect_uri, client)
-                except (ValidationError, InvalidRequestError):
+                    redirect_uri = client.validate_redirect_uri(raw_redirect_uri)
+                except (ValidationError, InvalidRedirectUriError):
                     pass
             if state is None:
                 # make last-ditch effort to load state
@@ -213,22 +184,22 @@ async def error_response(
 
             # Validate redirect_uri against client's registered URIs
             try:
-                redirect_uri = validate_redirect_uri(auth_request.redirect_uri, client)
-            except InvalidRequestError as validation_error:
+                redirect_uri = client.validate_redirect_uri(auth_request.redirect_uri)
+            except InvalidRedirectUriError as validation_error:
                 # For redirect_uri validation errors, return direct error (no redirect)
                 return await error_response(
                     error="invalid_request",
-                    error_description=validation_error.error_description,
+                    error_description=validation_error.message,
                 )
 
             # Validate scope - for scope errors, we can redirect
             try:
-                scopes = validate_scope(auth_request.scope, client)
-            except InvalidRequestError as validation_error:
+                scopes = client.validate_scope(auth_request.scope)
+            except InvalidScopeError as validation_error:
                 # For scope errors, redirect with error parameters
                 return await error_response(
                     error="invalid_scope",
-                    error_description=validation_error.error_description,
+                    error_description=validation_error.message,
                 )
 
             # Setup authorization parameters

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

@@ -7,11 +7,11 @@
 from starlette.responses import Response
 
 from mcp.server.auth.errors import (
-    InvalidClientError,
     stringify_pydantic_error,
 )
 from mcp.server.auth.json_response import PydanticJSONResponse
 from mcp.server.auth.middleware.client_auth import (
+    AuthenticationError,
     ClientAuthenticator,
 )
 from mcp.server.auth.provider import AuthInfo, OAuthServerProvider, RefreshToken
@@ -29,7 +29,7 @@ class RevocationRequest(BaseModel):
 
 
 class RevocationErrorResponse(BaseModel):
-    error: Literal["invalid_request",]
+    error: Literal["invalid_request", "unauthorized_client"]
     error_description: str | None = None
 
 
@@ -59,8 +59,14 @@ async def handle(self, request: Request) -> Response:
             client = await self.client_authenticator.authenticate(
                 revocation_request.client_id, revocation_request.client_secret
             )
-        except InvalidClientError as e:
-            return PydanticJSONResponse(status_code=401, content=e.error_response())
+        except AuthenticationError as e:
+            return PydanticJSONResponse(
+                status_code=401,
+                content=RevocationErrorResponse(
+                    error="unauthorized_client",
+                    error_description=e.message,
+                ),
+            )
 
         loaders = [
             self.provider.load_access_token,

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

@@ -9,11 +9,11 @@
 
 from mcp.server.auth.errors import (
     ErrorResponse,
-    InvalidClientError,
     stringify_pydantic_error,
 )
 from mcp.server.auth.json_response import PydanticJSONResponse
 from mcp.server.auth.middleware.client_auth import (
+    AuthenticationError,
     ClientAuthenticator,
 )
 from mcp.server.auth.provider import OAuthServerProvider
@@ -111,8 +111,13 @@ async def handle(self, request: Request):
                 client_id=token_request.client_id,
                 client_secret=token_request.client_secret,
             )
-        except InvalidClientError as e:
-            return self.response(e.error_response())
+        except AuthenticationError as e:
+            return self.response(
+                TokenErrorResponse(
+                    error="unauthorized_client",
+                    error_description=e.message,
+                )
+            )
 
         if token_request.grant_type not in client_info.grant_types:
             return self.response(

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

@@ -1,10 +1,14 @@
 import time
 
-from mcp.server.auth.errors import InvalidClientError
 from mcp.server.auth.provider import OAuthRegisteredClientsStore
 from mcp.shared.auth import OAuthClientInformationFull
 
 
+class AuthenticationError(Exception):
+    def __init__(self, message: str):
+        self.message = message
+
+
 class ClientAuthenticator:
     """
     ClientAuthenticator is a callable which validates requests from a client
@@ -31,21 +35,21 @@ async def authenticate(
         # Look up client information
         client = await self.clients_store.get_client(client_id)
         if not client:
-            raise InvalidClientError("Invalid client_id")
+            raise AuthenticationError("Invalid client_id")
 
         # If client from the store expects a secret, validate that the request provides
         # that secret
         if client.client_secret:
             if not client_secret:
-                raise InvalidClientError("Client secret is required")
+                raise AuthenticationError("Client secret is required")
 
             if client.client_secret != client_secret:
-                raise InvalidClientError("Invalid client_secret")
+                raise AuthenticationError("Invalid client_secret")
 
             if (
                 client.client_secret_expires_at
                 and client.client_secret_expires_at < int(time.time())
             ):
-                raise InvalidClientError("Client secret has expired")
+                raise AuthenticationError("Client secret has expired")
 
         return client

src/mcp/shared/auth.py

@@ -15,6 +15,16 @@ class OAuthToken(BaseModel):
     refresh_token: str | None = None
 
 
+class InvalidScopeError(Exception):
+    def __init__(self, message: str):
+        self.message = message
+
+
+class InvalidRedirectUriError(Exception):
+    def __init__(self, message: str):
+        self.message = message
+
+
 class OAuthClientMetadata(BaseModel):
     """
     RFC 7591 OAuth 2.0 Dynamic Client Registration metadata.
@@ -50,6 +60,32 @@ class OAuthClientMetadata(BaseModel):
     software_id: str | None = None
     software_version: str | None = None
 
+    def validate_scope(self, requested_scope: str | None) -> list[str] | None:
+        if requested_scope is None:
+            return None
+        requested_scopes = requested_scope.split(" ")
+        allowed_scopes = [] if self.scope is None else self.scope.split(" ")
+        for scope in requested_scopes:
+            if scope not in allowed_scopes:
+                raise InvalidScopeError(f"Client was not registered with scope {scope}")
+        return requested_scopes
+
+    def validate_redirect_uri(self, redirect_uri: AnyHttpUrl | None) -> AnyHttpUrl:
+        if redirect_uri is not None:
+            # Validate redirect_uri against client's registered redirect URIs
+            if redirect_uri not in self.redirect_uris:
+                raise InvalidRedirectUriError(
+                    f"Redirect URI '{redirect_uri}' not registered for client"
+                )
+            return redirect_uri
+        elif len(self.redirect_uris) == 1:
+            return self.redirect_uris[0]
+        else:
+            raise InvalidRedirectUriError(
+                "redirect_uri must be specified when client "
+                "has multiple registered URIs"
+            )
+
 
 class OAuthClientInformationFull(OAuthClientMetadata):
     """

tests/server/fastmcp/auth/test_auth_integration.py

@@ -198,7 +198,7 @@ async def load_access_token(self, token: str) -> AuthInfo | None:
             expires_at=token_info.expires_at,
         )
 
-    async def revoke_token(self, token: OAuthToken | RefreshToken) -> None:
+    async def revoke_token(self, token: AuthInfo | RefreshToken) -> None:
         match token:
             case RefreshToken():
                 # Remove the refresh token