feat: Follow OAuth 2.1 spec requirements on auth failures (#1923)

Co-authored-by: Tomas <>

Author: tcarac

src/fastmcp/server/auth/oauth_proxy.py

@@ -42,7 +42,7 @@
 from mcp.shared.auth import OAuthClientInformationFull, OAuthToken
 from pydantic import AnyHttpUrl, AnyUrl, SecretStr
 from starlette.requests import Request
-from starlette.responses import RedirectResponse
+from starlette.responses import JSONResponse, RedirectResponse
 from starlette.routing import Route
 
 import fastmcp
@@ -844,16 +844,25 @@ def get_routes(
                 f"Route {i}: {route} - path: {getattr(route, 'path', 'N/A')}, methods: {getattr(route, 'methods', 'N/A')}"
             )
 
-            # Keep all standard OAuth routes unchanged - our DCR-compliant flow handles everything
-            custom_routes.append(route)
-
+            # Replace the token endpoint with our custom handler that returns proper OAuth 2.1 error codes
             if (
                 isinstance(route, Route)
                 and route.path == "/token"
                 and route.methods is not None
                 and "POST" in route.methods
             ):
                 token_route_found = True
+                # Replace with our custom token handler
+                custom_routes.append(
+                    Route(
+                        path="/token",
+                        endpoint=self._handle_token_request,
+                        methods=["POST"],
+                    )
+                )
+            else:
+                # Keep all other standard OAuth routes unchanged
+                custom_routes.append(route)
 
         # Add OAuth callback endpoint for forwarding to client callbacks
         custom_routes.append(
@@ -869,6 +878,81 @@ def get_routes(
         )
         return custom_routes
 
+    # -------------------------------------------------------------------------
+    # Custom Token Endpoint Handler
+    # -------------------------------------------------------------------------
+
+    async def _handle_token_request(self, request: Request) -> JSONResponse:
+        """Handle token requests with proper OAuth 2.1 error handling.
+
+        This custom handler wraps the standard MCP SDK token handler but provides
+        OAuth 2.1 compliant error responses for client authentication failures:
+        - Returns HTTP 401 status code for client authentication failures
+        - Uses 'invalid_client' error code instead of 'unauthorized_client'
+
+        Per OAuth 2.1 spec: "The authorization server MAY return an HTTP 401
+        (Unauthorized) status code to indicate which HTTP authentication schemes
+        are supported. If the client attempted to authenticate via the Authorization
+        request header field, the authorization server MUST respond with an HTTP 401
+        (Unauthorized) status code and include the WWW-Authenticate response header
+        field matching the authentication scheme used by the client."
+        """
+        from mcp.server.auth.handlers.token import TokenHandler
+        from mcp.server.auth.middleware.client_auth import ClientAuthenticator
+
+        # Create the standard token handler and client authenticator
+        token_handler = TokenHandler(
+            provider=self, client_authenticator=ClientAuthenticator(self)
+        )
+
+        # Handle the request normally
+        response = await token_handler.handle(request)
+
+        # Check if the response is an error response for client authentication failure
+        if (
+            hasattr(response, "body")
+            and hasattr(response, "status_code")
+            and response.status_code == 400
+        ):
+            try:
+                import json
+
+                # Parse the response body to check for client authentication errors
+                body_content = (
+                    response.body.decode("utf-8")
+                    if hasattr(response.body, "decode")
+                    else str(response.body)
+                )
+                error_data = json.loads(body_content)
+
+                # Check if this is an unauthorized_client error (which means invalid client_id)
+                if error_data.get(
+                    "error"
+                ) == "unauthorized_client" and "Invalid client_id" in str(
+                    error_data.get("error_description", "")
+                ):
+                    logger.debug(
+                        "Client authentication failed - client not found, returning OAuth 2.1 compliant error"
+                    )
+
+                    # Return the correct OAuth 2.1 response
+                    return JSONResponse(
+                        content={
+                            "error": "invalid_client",
+                            "error_description": error_data.get("error_description"),
+                        },
+                        status_code=401,
+                        headers={
+                            "Cache-Control": "no-store",
+                            "Pragma": "no-cache",
+                        },
+                    )
+            except (json.JSONDecodeError, AttributeError, KeyError):
+                # If we can't parse the response, return it as-is
+                pass
+
+        return response
+
     # -------------------------------------------------------------------------
     # IdP Callback Forwarding
     # -------------------------------------------------------------------------

tests/server/auth/test_oauth_proxy.py

@@ -973,3 +973,59 @@ async def test_multiple_extra_params(self, jwt_verifier):
         assert query_params["audience"][0] == "https://api.example.com"
         assert query_params["prompt"][0] == "consent"
         assert query_params["max_age"][0] == "3600"
+
+    @pytest.mark.asyncio
+    async def test_token_endpoint_invalid_client_error(self, jwt_verifier):
+        """Test that invalid client_id returns OAuth 2.1 compliant error response.
+
+        When a client ID is not found during token exchange, the proxy should:
+        1. Return HTTP 401 status code
+        2. Use 'invalid_client' error code instead of 'unauthorized_client'
+
+        This aligns with OAuth 2.1 spec and enables Claude's automatic client re-registration.
+        """
+        from starlette.applications import Starlette
+        from starlette.testclient import TestClient
+
+        proxy = OAuthProxy(
+            upstream_authorization_endpoint="https://oauth.example.com/authorize",
+            upstream_token_endpoint="https://oauth.example.com/token",
+            upstream_client_id="upstream-client",
+            upstream_client_secret="upstream-secret",
+            token_verifier=jwt_verifier,
+            base_url="https://proxy.example.com",
+        )
+
+        # Create a test app with OAuth routes
+        app = Starlette(routes=proxy.get_routes())
+
+        # Test the token endpoint with an invalid (non-existent) client_id
+        with TestClient(app) as client:
+            response = client.post(
+                "/token",
+                data={
+                    "grant_type": "authorization_code",
+                    "code": "test-auth-code",
+                    "client_id": "non-existent-client-id",
+                    "code_verifier": "test-code-verifier",
+                    "redirect_uri": "http://localhost:12345/callback",
+                },
+                headers={
+                    "Content-Type": "application/x-www-form-urlencoded",
+                },
+            )
+
+            # Verify OAuth 2.1 compliant error response
+            assert response.status_code == 401, (
+                f"Expected 401 but got {response.status_code}"
+            )
+
+            error_data = response.json()
+            assert error_data["error"] == "invalid_client", (
+                f"Expected 'invalid_client' but got '{error_data.get('error')}'"
+            )
+            assert "Invalid client_id" in error_data["error_description"]
+
+            # Verify proper cache headers are set
+            assert response.headers.get("Cache-Control") == "no-store"
+            assert response.headers.get("Pragma") == "no-cache"