From 562e51b37823be6e8430f85a5bbddc2d042ff4c9 Mon Sep 17 00:00:00 2001
From: Jeremiah Lowin <153965+jlowin@users.noreply.github.com>
Date: Wed, 22 Oct 2025 21:29:14 -0400
Subject: [PATCH] Improve OAuth error messages with custom handlers and
middleware (#2221)
* Enhance OAuth Proxy error responses with branded HTML pages
OAuth Proxy authentication errors now show styled HTML error pages in browsers
instead of raw JSON, with content negotiation for API clients. Enhanced error
messages explain common causes (ephemeral storage, server restarts) and provide
clear remediation steps.
Changes:
- Created enhanced authorization handler that extends SDK's AuthorizationHandler
- Created enhanced auth middleware that extends SDK's RequireAuthMiddleware
- HTML error pages use server branding (icon, name) from FastMCP instance
- Added comprehensive troubleshooting section to OAuth Proxy docs
- Added FAQ entry linking to detailed troubleshooting
* Add comprehensive tests for enhanced OAuth error responses
Tests cover:
- HTML error pages for browser requests with server branding
- Enhanced JSON responses with registration endpoint hints
- Content negotiation between HTML and JSON
- Enhanced middleware error messages for invalid_token
- WWW-Authenticate header format consistency with SDK
* Update language for new storage defaults
* update docs
* Update tests for simplified error messages
* Clean up messages
---
src/fastmcp/server/auth/handlers/authorize.py | 324 ++++++++++++++++
src/fastmcp/server/auth/middleware.py | 96 +++++
src/fastmcp/server/auth/oauth_proxy.py | 35 +-
src/fastmcp/server/http.py | 2 +-
.../auth/test_enhanced_error_responses.py | 362 ++++++++++++++++++
5 files changed, 813 insertions(+), 6 deletions(-)
create mode 100644 src/fastmcp/server/auth/handlers/authorize.py
create mode 100644 src/fastmcp/server/auth/middleware.py
create mode 100644 tests/server/auth/test_enhanced_error_responses.py
diff --git a/src/fastmcp/server/auth/handlers/authorize.py b/src/fastmcp/server/auth/handlers/authorize.py
new file mode 100644
index 000000000..13c98a56d
--- /dev/null
+++ b/src/fastmcp/server/auth/handlers/authorize.py
@@ -0,0 +1,324 @@
+"""Enhanced authorization handler with improved error responses.
+
+This module provides an enhanced authorization handler that wraps the MCP SDK's
+AuthorizationHandler to provide better error messages when clients attempt to
+authorize with unregistered client IDs.
+
+The enhancement adds:
+- Content negotiation: HTML for browsers, JSON for API clients
+- Enhanced JSON responses with registration endpoint hints
+- Styled HTML error pages with registration links/forms
+- Link headers pointing to registration endpoints
+"""
+
+from __future__ import annotations
+
+from typing import TYPE_CHECKING
+
+from mcp.server.auth.handlers.authorize import (
+ AuthorizationHandler as SDKAuthorizationHandler,
+)
+from pydantic import AnyHttpUrl
+from starlette.requests import Request
+from starlette.responses import Response
+
+from fastmcp.utilities.logging import get_logger
+from fastmcp.utilities.ui import (
+ INFO_BOX_STYLES,
+ TOOLTIP_STYLES,
+ create_logo,
+ create_page,
+ create_secure_html_response,
+)
+
+if TYPE_CHECKING:
+ from mcp.server.auth.provider import OAuthAuthorizationServerProvider
+
+logger = get_logger(__name__)
+
+
+def create_unregistered_client_html(
+ client_id: str,
+ registration_endpoint: str,
+ discovery_endpoint: str,
+ server_name: str | None = None,
+ server_icon_url: str | None = None,
+ title: str = "Client Not Registered",
+) -> str:
+ """Create styled HTML error page for unregistered client attempts.
+
+ Args:
+ client_id: The unregistered client ID that was provided
+ registration_endpoint: URL of the registration endpoint
+ discovery_endpoint: URL of the OAuth metadata discovery endpoint
+ server_name: Optional server name for branding
+ server_icon_url: Optional server icon URL
+ title: Page title
+
+ Returns:
+ HTML string for the error page
+ """
+ import html as html_module
+
+ client_id_escaped = html_module.escape(client_id)
+
+ # Main error message
+ error_box = f"""
+
+
The client ID {client_id_escaped} was not found in the server's client registry.
+
+
Your MCP client opened this page to complete OAuth authorization,
+ but the server did not recognize its client ID. To fix this:
+
+ - Close this browser window
+ - Clear authentication tokens in your MCP client (or restart it)
+ - Try connecting again - your client should automatically re-register
+
+
+
+ Why am I seeing this?
+
+ OAuth 2.0 requires clients to register before authorization.
+ This server returned a 400 error because the provided client
+ ID was not found.
+
+ In browser-delegated OAuth flows, your application cannot
+ detect this error automatically; it's waiting for a
+ callback that will never arrive. You must manually clear
+ auth tokens and reconnect.
+
+
+
+ """
+
+ # Build page content
+ content = f"""
+
+ {create_logo(icon_url=server_icon_url, alt_text=server_name or "FastMCP")}
+
{title}
+ {error_box}
+ {warning_box}
+
+ {help_link}
+ """
+
+ # Use same styles as consent page
+ additional_styles = (
+ INFO_BOX_STYLES
+ + TOOLTIP_STYLES
+ + """
+ /* Error variant for info-box */
+ .info-box.error {
+ background: #fef2f2;
+ border-color: #f87171;
+ }
+ .info-box.error strong {
+ color: #991b1b;
+ }
+ /* Warning variant for info-box (yellow) */
+ .info-box.warning {
+ background: #fffbeb;
+ border-color: #fbbf24;
+ }
+ .info-box.warning strong {
+ color: #92400e;
+ }
+ .info-box code {
+ background: rgba(0, 0, 0, 0.05);
+ padding: 2px 6px;
+ border-radius: 3px;
+ font-family: 'SF Mono', Monaco, 'Cascadia Code', monospace;
+ font-size: 0.9em;
+ }
+ .info-box ul {
+ margin: 10px 0;
+ padding-left: 20px;
+ }
+ .info-box li {
+ margin: 6px 0;
+ }
+ """
+ )
+
+ return create_page(
+ content=content,
+ title=title,
+ additional_styles=additional_styles,
+ )
+
+
+class AuthorizationHandler(SDKAuthorizationHandler):
+ """Authorization handler with enhanced error responses for unregistered clients.
+
+ This handler extends the MCP SDK's AuthorizationHandler to provide better UX
+ when clients attempt to authorize without being registered. It implements
+ content negotiation to return:
+
+ - HTML error pages for browser requests
+ - Enhanced JSON with registration hints for API clients
+ - Link headers pointing to registration endpoints
+
+ This maintains OAuth 2.1 compliance (returns 400 for invalid client_id)
+ while providing actionable guidance to fix the error.
+ """
+
+ def __init__(
+ self,
+ provider: OAuthAuthorizationServerProvider,
+ base_url: AnyHttpUrl | str,
+ server_name: str | None = None,
+ server_icon_url: str | None = None,
+ ):
+ """Initialize the enhanced authorization handler.
+
+ Args:
+ provider: OAuth authorization server provider
+ base_url: Base URL of the server for constructing endpoint URLs
+ server_name: Optional server name for branding
+ server_icon_url: Optional server icon URL for branding
+ """
+ super().__init__(provider)
+ self._base_url = str(base_url).rstrip("/")
+ self._server_name = server_name
+ self._server_icon_url = server_icon_url
+
+ async def handle(self, request: Request) -> Response:
+ """Handle authorization request with enhanced error responses.
+
+ This method extends the SDK's authorization handler and intercepts
+ errors for unregistered clients to provide better error responses
+ based on the client's Accept header.
+
+ Args:
+ request: The authorization request
+
+ Returns:
+ Response (redirect on success, error response on failure)
+ """
+ # Call the SDK handler
+ response = await super().handle(request)
+
+ # Check if this is a client not found error
+ if response.status_code == 400:
+ # Try to extract client_id from request for enhanced error
+ client_id = None
+ if request.method == "GET":
+ client_id = request.query_params.get("client_id")
+ else:
+ form = await request.form()
+ client_id = form.get("client_id")
+
+ # If we have a client_id and the error is about it not being found,
+ # enhance the response
+ if client_id:
+ try:
+ # Check if response body contains "not found" error
+ if hasattr(response, "body"):
+ import json
+
+ body = json.loads(response.body)
+ if (
+ body.get("error") == "invalid_request"
+ and "not found" in body.get("error_description", "").lower()
+ ):
+ return await self._create_enhanced_error_response(
+ request, client_id, body.get("state")
+ )
+ except Exception:
+ # If we can't parse the response, just return the original
+ pass
+
+ return response
+
+ async def _create_enhanced_error_response(
+ self, request: Request, client_id: str, state: str | None
+ ) -> Response:
+ """Create enhanced error response with content negotiation.
+
+ Args:
+ request: The original request
+ client_id: The unregistered client ID
+ state: The state parameter from the request
+
+ Returns:
+ HTML or JSON error response based on Accept header
+ """
+ registration_endpoint = f"{self._base_url}/register"
+ discovery_endpoint = f"{self._base_url}/.well-known/oauth-authorization-server"
+
+ # Extract server metadata from app state (same pattern as consent screen)
+ from fastmcp.server.server import FastMCP
+
+ fastmcp = getattr(request.app.state, "fastmcp_server", None)
+
+ if isinstance(fastmcp, FastMCP):
+ server_name = fastmcp.name
+ icons = fastmcp.icons
+ server_icon_url = icons[0].src if icons else None
+ else:
+ server_name = self._server_name
+ server_icon_url = self._server_icon_url
+
+ # Check Accept header for content negotiation
+ accept = request.headers.get("accept", "")
+
+ # Prefer HTML for browsers
+ if "text/html" in accept:
+ html = create_unregistered_client_html(
+ client_id=client_id,
+ registration_endpoint=registration_endpoint,
+ discovery_endpoint=discovery_endpoint,
+ server_name=server_name,
+ server_icon_url=server_icon_url,
+ )
+ response = create_secure_html_response(html, status_code=400)
+ else:
+ # Return enhanced JSON for API clients
+ from mcp.server.auth.handlers.authorize import AuthorizationErrorResponse
+
+ error_data = AuthorizationErrorResponse(
+ error="invalid_request",
+ error_description=(
+ f"Client ID '{client_id}' is not registered with this server. "
+ f"MCP clients should automatically re-register by sending a POST request to "
+ f"the registration_endpoint and retry authorization. "
+ f"If this persists, clear cached authentication tokens and reconnect."
+ ),
+ state=state,
+ )
+
+ # Add extra fields to help clients discover registration
+ error_dict = error_data.model_dump(exclude_none=True)
+ error_dict["registration_endpoint"] = registration_endpoint
+ error_dict["authorization_server_metadata"] = discovery_endpoint
+
+ from starlette.responses import JSONResponse
+
+ response = JSONResponse(
+ status_code=400,
+ content=error_dict,
+ headers={"Cache-Control": "no-store"},
+ )
+
+ # Add Link header for registration endpoint discovery
+ response.headers["Link"] = (
+ f'<{registration_endpoint}>; rel="http://oauth.net/core/2.1/#registration"'
+ )
+
+ logger.info(
+ "Unregistered client_id=%s, returned %s error response",
+ client_id,
+ "HTML" if "text/html" in accept else "JSON",
+ )
+
+ return response
diff --git a/src/fastmcp/server/auth/middleware.py b/src/fastmcp/server/auth/middleware.py
new file mode 100644
index 000000000..a7e80a7ff
--- /dev/null
+++ b/src/fastmcp/server/auth/middleware.py
@@ -0,0 +1,96 @@
+"""Enhanced authentication middleware with better error messages.
+
+This module provides enhanced versions of MCP SDK authentication middleware
+that return more helpful error messages for developers troubleshooting
+authentication issues.
+"""
+
+from __future__ import annotations
+
+import json
+
+from mcp.server.auth.middleware.bearer_auth import (
+ RequireAuthMiddleware as SDKRequireAuthMiddleware,
+)
+from starlette.types import Send
+
+from fastmcp.utilities.logging import get_logger
+
+logger = get_logger(__name__)
+
+
+class RequireAuthMiddleware(SDKRequireAuthMiddleware):
+ """Enhanced authentication middleware with detailed error messages.
+
+ Extends the SDK's RequireAuthMiddleware to provide more actionable
+ error messages when authentication fails. This helps developers
+ understand what went wrong and how to fix it.
+ """
+
+ async def _send_auth_error(
+ self, send: Send, status_code: int, error: str, description: str
+ ) -> None:
+ """Send an authentication error response with enhanced error messages.
+
+ Overrides the SDK's _send_auth_error to provide more detailed
+ error descriptions that help developers troubleshoot authentication
+ issues.
+
+ Args:
+ send: ASGI send callable
+ status_code: HTTP status code (401 or 403)
+ error: OAuth error code
+ description: Base error description
+ """
+ # Enhance error descriptions based on error type
+ enhanced_description = description
+
+ if error == "invalid_token" and status_code == 401:
+ # This is the "Authentication required" error
+ enhanced_description = (
+ "Authentication failed. The provided bearer token is invalid, expired, or no longer recognized by the server. "
+ "To resolve: clear authentication tokens in your MCP client and reconnect. "
+ "Your client should automatically re-register and obtain new tokens."
+ )
+ elif error == "insufficient_scope":
+ # Scope error - already has good detail from SDK
+ pass
+
+ # Build WWW-Authenticate header value
+ www_auth_parts = [
+ f'error="{error}"',
+ f'error_description="{enhanced_description}"',
+ ]
+ if self.resource_metadata_url:
+ www_auth_parts.append(f'resource_metadata="{self.resource_metadata_url}"')
+
+ www_authenticate = f"Bearer {', '.join(www_auth_parts)}"
+
+ # Send response
+ body = {"error": error, "error_description": enhanced_description}
+ body_bytes = json.dumps(body).encode()
+
+ await send(
+ {
+ "type": "http.response.start",
+ "status": status_code,
+ "headers": [
+ (b"content-type", b"application/json"),
+ (b"content-length", str(len(body_bytes)).encode()),
+ (b"www-authenticate", www_authenticate.encode()),
+ ],
+ }
+ )
+
+ await send(
+ {
+ "type": "http.response.body",
+ "body": body_bytes,
+ }
+ )
+
+ logger.info(
+ "Auth error returned: %s (status=%d)",
+ error,
+ status_code,
+ )
diff --git a/src/fastmcp/server/auth/oauth_proxy.py b/src/fastmcp/server/auth/oauth_proxy.py
index 9088a6659..aaa06be56 100644
--- a/src/fastmcp/server/auth/oauth_proxy.py
+++ b/src/fastmcp/server/auth/oauth_proxy.py
@@ -60,6 +60,7 @@ from starlette.routing import Route
from fastmcp import settings
from fastmcp.server.auth.auth import OAuthProvider, TokenVerifier
+from fastmcp.server.auth.handlers.authorize import AuthorizationHandler
from fastmcp.server.auth.jwt_issuer import (
JWTIssuer,
TokenEncryption,
@@ -1590,10 +1591,11 @@ class OAuthProxy(OAuthProvider):
self,
mcp_path: str | None = None,
) -> list[Route]:
- """Get OAuth routes with custom proxy token handler.
+ """Get OAuth routes with custom handlers for better error UX.
- This method creates standard OAuth routes and replaces the token endpoint
- with our proxy handler that forwards requests to the upstream OAuth server.
+ This method creates standard OAuth routes and replaces:
+ - /authorize endpoint: Enhanced error responses for unregistered clients
+ - /token endpoint: OAuth 2.1 compliant error codes
Args:
mcp_path: The path where the MCP endpoint is mounted (e.g., "/mcp")
@@ -1603,6 +1605,7 @@ class OAuthProxy(OAuthProvider):
routes = super().get_routes(mcp_path)
custom_routes = []
token_route_found = False
+ authorize_route_found = False
logger.debug(
f"get_routes called - configuring OAuth routes in {len(routes)} routes"
@@ -1613,8 +1616,30 @@ class OAuthProxy(OAuthProvider):
f"Route {i}: {route} - path: {getattr(route, 'path', 'N/A')}, methods: {getattr(route, 'methods', 'N/A')}"
)
- # Replace the token endpoint with our custom handler that returns proper OAuth 2.1 error codes
+ # Replace the authorize endpoint with our enhanced handler for better error UX
if (
+ isinstance(route, Route)
+ and route.path == "/authorize"
+ and route.methods is not None
+ and ("GET" in route.methods or "POST" in route.methods)
+ ):
+ authorize_route_found = True
+ # Replace with our enhanced authorization handler
+ authorize_handler = AuthorizationHandler(
+ provider=self,
+ base_url=self.base_url,
+ server_name=None, # Could be extended to pass server metadata
+ server_icon_url=None,
+ )
+ custom_routes.append(
+ Route(
+ path="/authorize",
+ endpoint=authorize_handler.handle,
+ methods=["GET", "POST"],
+ )
+ )
+ # Replace the token endpoint with our custom handler that returns proper OAuth 2.1 error codes
+ elif (
isinstance(route, Route)
and route.path == "/token"
and route.methods is not None
@@ -1658,7 +1683,7 @@ class OAuthProxy(OAuthProvider):
)
logger.debug(
- f"✅ OAuth routes configured: token_endpoint={token_route_found}, total routes={len(custom_routes)} (includes OAuth callback + consent)"
+ f"✅ OAuth routes configured: authorize_endpoint={authorize_route_found}, token_endpoint={token_route_found}, total routes={len(custom_routes)} (includes OAuth callback + consent)"
)
return custom_routes
diff --git a/src/fastmcp/server/http.py b/src/fastmcp/server/http.py
index d7c3bc698..2ac186761 100644
--- a/src/fastmcp/server/http.py
+++ b/src/fastmcp/server/http.py
@@ -5,7 +5,6 @@ from contextlib import asynccontextmanager, contextmanager
from contextvars import ContextVar
from typing import TYPE_CHECKING
-from mcp.server.auth.middleware.bearer_auth import RequireAuthMiddleware
from mcp.server.auth.routes import build_resource_metadata_url
from mcp.server.lowlevel.server import LifespanResultT
from mcp.server.sse import SseServerTransport
@@ -21,6 +20,7 @@ from starlette.routing import BaseRoute, Mount, Route
from starlette.types import Lifespan, Receive, Scope, Send
from fastmcp.server.auth import AuthProvider
+from fastmcp.server.auth.middleware import RequireAuthMiddleware
from fastmcp.utilities.logging import get_logger
if TYPE_CHECKING:
diff --git a/tests/server/auth/test_enhanced_error_responses.py b/tests/server/auth/test_enhanced_error_responses.py
new file mode 100644
index 000000000..4f345b38d
--- /dev/null
+++ b/tests/server/auth/test_enhanced_error_responses.py
@@ -0,0 +1,362 @@
+"""Tests for enhanced OAuth error responses.
+
+This test suite covers:
+1. Enhanced authorization handler (HTML and JSON error pages)
+2. Enhanced middleware (better error messages)
+3. Content negotiation
+4. Server branding in error pages
+"""
+
+import pytest
+from mcp.shared.auth import OAuthClientInformationFull
+from pydantic import AnyUrl
+from starlette.applications import Starlette
+from starlette.testclient import TestClient
+
+from fastmcp import FastMCP
+from fastmcp.server.auth.oauth_proxy import OAuthProxy
+from fastmcp.server.auth.providers.jwt import JWTVerifier, RSAKeyPair
+
+
+class TestEnhancedAuthorizationHandler:
+ """Tests for enhanced authorization handler error responses."""
+
+ @pytest.fixture
+ def rsa_key_pair(self) -> RSAKeyPair:
+ """Generate RSA key pair for testing."""
+ return RSAKeyPair.generate()
+
+ @pytest.fixture
+ def oauth_proxy(self, rsa_key_pair):
+ """Create OAuth proxy for testing."""
+ return OAuthProxy(
+ upstream_authorization_endpoint="https://github.com/login/oauth/authorize",
+ upstream_token_endpoint="https://github.com/login/oauth/access_token",
+ upstream_client_id="test-client-id",
+ upstream_client_secret="test-client-secret",
+ token_verifier=JWTVerifier(
+ public_key=rsa_key_pair.public_key,
+ issuer="https://test.com",
+ audience="https://test.com",
+ base_url="https://test.com",
+ ),
+ base_url="https://myserver.com",
+ )
+
+ def test_unregistered_client_returns_html_for_browser(self, oauth_proxy):
+ """Test that unregistered client returns styled HTML for browser requests."""
+ app = Starlette(routes=oauth_proxy.get_routes())
+
+ with TestClient(app) as client:
+ response = client.get(
+ "/authorize",
+ params={
+ "client_id": "unregistered-client-id",
+ "redirect_uri": "http://localhost:12345/callback",
+ "response_type": "code",
+ "code_challenge": "test-challenge",
+ "state": "test-state",
+ },
+ headers={"Accept": "text/html"},
+ )
+
+ # Should return 400 with HTML content
+ assert response.status_code == 400
+ assert "text/html" in response.headers["content-type"]
+
+ # HTML should contain error message
+ html = response.text
+ assert "Client Not Registered" in html
+ assert "unregistered-client-id" in html
+ assert "To fix this" in html
+ assert "Close this browser window" in html
+ assert "Clear authentication tokens" in html
+
+ # Should have Link header for registration endpoint
+ assert "Link" in response.headers
+ assert "/register" in response.headers["Link"]
+
+ def test_unregistered_client_returns_json_for_api(self, oauth_proxy):
+ """Test that unregistered client returns enhanced JSON for API clients."""
+ app = Starlette(routes=oauth_proxy.get_routes())
+
+ with TestClient(app) as client:
+ response = client.get(
+ "/authorize",
+ params={
+ "client_id": "unregistered-client-id",
+ "redirect_uri": "http://localhost:12345/callback",
+ "response_type": "code",
+ "code_challenge": "test-challenge",
+ "state": "test-state",
+ },
+ headers={"Accept": "application/json"},
+ )
+
+ # Should return 400 with JSON content
+ assert response.status_code == 400
+ assert "application/json" in response.headers["content-type"]
+
+ # JSON should have enhanced error response
+ data = response.json()
+ assert data["error"] == "invalid_request"
+ assert "unregistered-client-id" in data["error_description"]
+ assert data["state"] == "test-state"
+
+ # Should include registration endpoint hints
+ assert "registration_endpoint" in data
+ assert data["registration_endpoint"] == "https://myserver.com/register"
+ assert "authorization_server_metadata" in data
+
+ # Should have Link header
+ assert "Link" in response.headers
+ assert "/register" in response.headers["Link"]
+
+ def test_successful_authorization_not_enhanced(self, oauth_proxy):
+ """Test that successful authorizations are not modified by enhancement."""
+ app = Starlette(routes=oauth_proxy.get_routes())
+
+ # Register a valid client first
+ client_info = OAuthClientInformationFull(
+ client_id="valid-client",
+ client_secret="valid-secret",
+ redirect_uris=[AnyUrl("http://localhost:12345/callback")],
+ )
+
+ # Need to register synchronously
+ import asyncio
+
+ asyncio.run(oauth_proxy.register_client(client_info))
+
+ with TestClient(app) as client:
+ response = client.get(
+ "/authorize",
+ params={
+ "client_id": "valid-client",
+ "redirect_uri": "http://localhost:12345/callback",
+ "response_type": "code",
+ "code_challenge": "test-challenge",
+ "state": "test-state",
+ },
+ headers={"Accept": "text/html"},
+ follow_redirects=False,
+ )
+
+ # Should redirect to consent page (302), not return error
+ assert response.status_code == 302
+ assert "/consent" in response.headers["location"]
+
+ def test_html_error_includes_server_branding(self, oauth_proxy):
+ """Test that HTML error page includes server branding from FastMCP instance."""
+ from mcp.types import Icon
+
+ # Create FastMCP server with custom branding
+ mcp = FastMCP(
+ "My Custom Server",
+ icons=[Icon(src="https://example.com/icon.png", mimeType="image/png")],
+ )
+
+ # Create app with OAuth routes
+ app = Starlette(routes=oauth_proxy.get_routes())
+ # Attach FastMCP instance to app state (same as done in http.py)
+ app.state.fastmcp_server = mcp
+
+ with TestClient(app) as client:
+ response = client.get(
+ "/authorize",
+ params={
+ "client_id": "unregistered-client-id",
+ "redirect_uri": "http://localhost:12345/callback",
+ "response_type": "code",
+ "code_challenge": "test-challenge",
+ },
+ headers={"Accept": "text/html"},
+ )
+
+ assert response.status_code == 400
+ html = response.text
+
+ # Should include custom server icon
+ assert "https://example.com/icon.png" in html
+
+
+class TestEnhancedRequireAuthMiddleware:
+ """Tests for enhanced authentication middleware error messages."""
+
+ @pytest.fixture
+ def rsa_key_pair(self) -> RSAKeyPair:
+ """Generate RSA key pair for testing."""
+ return RSAKeyPair.generate()
+
+ @pytest.fixture
+ def jwt_verifier(self, rsa_key_pair):
+ """Create JWT verifier for testing."""
+ return JWTVerifier(
+ public_key=rsa_key_pair.public_key,
+ issuer="https://test.com",
+ audience="https://test.com",
+ base_url="https://test.com",
+ )
+
+ def test_invalid_token_enhanced_error_message(self, jwt_verifier):
+ """Test that invalid_token errors have enhanced error messages."""
+ from fastmcp.server.http import create_streamable_http_app
+
+ server = FastMCP("Test Server")
+
+ @server.tool
+ def test_tool() -> str:
+ return "test"
+
+ app = create_streamable_http_app(
+ server=server,
+ streamable_http_path="/mcp",
+ auth=jwt_verifier,
+ )
+
+ with TestClient(app) as client:
+ # Request without Authorization header
+ response = client.post("/mcp")
+
+ assert response.status_code == 401
+ assert "www-authenticate" in response.headers
+
+ # Check enhanced error message
+ data = response.json()
+ assert data["error"] == "invalid_token"
+ # Should have enhanced description with resolution steps
+ assert "clear authentication tokens" in data["error_description"]
+ assert "automatically re-register" in data["error_description"]
+
+ def test_invalid_token_www_authenticate_header_format(self, jwt_verifier):
+ """Test that WWW-Authenticate header format matches SDK."""
+ from fastmcp.server.http import create_streamable_http_app
+
+ server = FastMCP("Test Server")
+ app = create_streamable_http_app(
+ server=server,
+ streamable_http_path="/mcp",
+ auth=jwt_verifier,
+ )
+
+ with TestClient(app) as client:
+ response = client.post("/mcp")
+
+ assert response.status_code == 401
+ www_auth = response.headers["www-authenticate"]
+
+ # Should follow Bearer challenge format
+ assert www_auth.startswith("Bearer ")
+ assert 'error="invalid_token"' in www_auth
+ assert "error_description=" in www_auth
+
+ def test_insufficient_scope_not_enhanced(self, rsa_key_pair):
+ """Test that insufficient_scope errors are not modified."""
+ # Create a valid token with wrong scopes
+ from fastmcp.server.http import create_streamable_http_app
+
+ jwt_verifier = JWTVerifier(
+ public_key=rsa_key_pair.public_key,
+ issuer="https://test.com",
+ audience="https://test.com",
+ base_url="https://test.com",
+ )
+
+ server = FastMCP("Test Server")
+
+ @server.tool
+ def test_tool() -> str:
+ return "test"
+
+ app = create_streamable_http_app(
+ server=server,
+ streamable_http_path="/mcp",
+ auth=jwt_verifier,
+ )
+
+ # Note: Testing insufficient_scope would require mocking the verifier
+ # to return a token with wrong scopes. For now, we verify the middleware
+ # is properly in place by checking it rejects unauthenticated requests.
+ with TestClient(app) as client:
+ response = client.post("/mcp")
+ # Without a valid token, we get invalid_token
+ assert response.status_code == 401
+
+
+class TestContentNegotiation:
+ """Tests for content negotiation in error responses."""
+
+ @pytest.fixture
+ def oauth_proxy(self):
+ """Create OAuth proxy for testing."""
+ return OAuthProxy(
+ upstream_authorization_endpoint="https://github.com/login/oauth/authorize",
+ upstream_token_endpoint="https://github.com/login/oauth/access_token",
+ upstream_client_id="test-client-id",
+ upstream_client_secret="test-client-secret",
+ token_verifier=JWTVerifier(
+ public_key=RSAKeyPair.generate().public_key,
+ issuer="https://test.com",
+ audience="https://test.com",
+ base_url="https://test.com",
+ ),
+ base_url="https://myserver.com",
+ )
+
+ def test_html_preferred_when_both_accepted(self, oauth_proxy):
+ """Test that HTML is preferred when both text/html and application/json are accepted."""
+ app = Starlette(routes=oauth_proxy.get_routes())
+
+ with TestClient(app) as client:
+ response = client.get(
+ "/authorize",
+ params={
+ "client_id": "unregistered-client-id",
+ "redirect_uri": "http://localhost:12345/callback",
+ "response_type": "code",
+ "code_challenge": "test-challenge",
+ },
+ headers={"Accept": "text/html,application/json"},
+ )
+
+ # Should prefer HTML
+ assert response.status_code == 400
+ assert "text/html" in response.headers["content-type"]
+
+ def test_json_when_only_json_accepted(self, oauth_proxy):
+ """Test that JSON is returned when only application/json is accepted."""
+ app = Starlette(routes=oauth_proxy.get_routes())
+
+ with TestClient(app) as client:
+ response = client.get(
+ "/authorize",
+ params={
+ "client_id": "unregistered-client-id",
+ "redirect_uri": "http://localhost:12345/callback",
+ "response_type": "code",
+ "code_challenge": "test-challenge",
+ },
+ headers={"Accept": "application/json"},
+ )
+
+ assert response.status_code == 400
+ assert "application/json" in response.headers["content-type"]
+
+ def test_json_when_no_accept_header(self, oauth_proxy):
+ """Test that JSON is returned when no Accept header is provided."""
+ app = Starlette(routes=oauth_proxy.get_routes())
+
+ with TestClient(app) as client:
+ response = client.get(
+ "/authorize",
+ params={
+ "client_id": "unregistered-client-id",
+ "redirect_uri": "http://localhost:12345/callback",
+ "response_type": "code",
+ "code_challenge": "test-challenge",
+ },
+ )
+
+ # Without Accept header, should return JSON (API default)
+ assert response.status_code == 400
+ assert "application/json" in response.headers["content-type"]