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"]