Enhance OAuth 2.0 support and middleware functionality

- Added resource_server_url and resource_name to AuthSettings for improved metadata handling.
- Updated FastMCP to utilize resource metadata URL in RequireAuthMiddleware.
- Introduced get_oauth_protected_resource_metadata_url function for generating resource metadata URLs.
- Modified tests to validate new metadata endpoints and middleware behavior.

Signed-off-by: Xin Fu <xfu83@bloomberg.net>

Author: imfing

examples/servers/simple-auth/mcp_simple_auth/server.py

@@ -261,6 +261,8 @@ def create_simple_mcp_server(settings: ServerSettings) -> FastMCP:
 
     auth_settings = AuthSettings(
         issuer_url=settings.server_url,
+        resource_server_url=settings.server_url,
+        resource_name="Simple GitHub MCP Server",
         client_registration_options=ClientRegistrationOptions(
             enabled=True,
             valid_scopes=[settings.mcp_scope],

src/mcp/server/auth/routes.py

@@ -1,5 +1,6 @@
 from collections.abc import Awaitable, Callable
 from typing import Any
+from urllib.parse import urljoin
 
 from pydantic import AnyHttpUrl
 from starlette.middleware.cors import CORSMiddleware
@@ -223,3 +224,7 @@ def build_metadata(
         metadata.revocation_endpoint_auth_methods_supported = ["client_secret_post"]
 
     return metadata
+
+
+def get_oauth_protected_resource_metadata_url(server_url: AnyHttpUrl) -> AnyHttpUrl:
+    return AnyHttpUrl(urljoin(str(server_url), "/.well-known/oauth-protected-resource"))

src/mcp/server/fastmcp/server.py

@@ -695,9 +695,15 @@ async def handle_sse(scope: Scope, receive: Receive, send: Send):
         # Add auth endpoints if auth provider is configured
         if self._auth_server_provider:
             assert self.settings.auth
-            from mcp.server.auth.routes import create_auth_routes
+            from mcp.server.auth.routes import (
+                create_auth_routes,
+                get_oauth_protected_resource_metadata_url,
+            )
 
             required_scopes = self.settings.auth.required_scopes or []
+            resource_metadata_url = get_oauth_protected_resource_metadata_url(
+                self.settings.auth.resource_server_url
+            )
 
             middleware = [
                 # extract auth info from request (but do not require it)
@@ -723,20 +729,24 @@ async def handle_sse(scope: Scope, receive: Receive, send: Send):
                 )
             )
 
-        # When auth is not configured, we shouldn't require auth
-        if self._auth_server_provider:
             # Auth is enabled, wrap the endpoints with RequireAuthMiddleware
             routes.append(
                 Route(
                     self.settings.sse_path,
-                    endpoint=RequireAuthMiddleware(handle_sse, required_scopes),
+                    endpoint=RequireAuthMiddleware(
+                        handle_sse, required_scopes, str(resource_metadata_url)
+                    ),
                     methods=["GET"],
                 )
             )
             routes.append(
                 Mount(
                     self.settings.message_path,
-                    app=RequireAuthMiddleware(sse.handle_post_message, required_scopes),
+                    app=RequireAuthMiddleware(
+                        sse.handle_post_message,
+                        required_scopes,
+                        str(resource_metadata_url),
+                    ),
                 )
             )
         else:
@@ -795,9 +805,15 @@ async def handle_streamable_http(
         # Add auth endpoints if auth provider is configured
         if self._auth_server_provider:
             assert self.settings.auth
-            from mcp.server.auth.routes import create_auth_routes
+            from mcp.server.auth.routes import (
+                create_auth_routes,
+                get_oauth_protected_resource_metadata_url,
+            )
 
             required_scopes = self.settings.auth.required_scopes or []
+            resource_metadata_url = get_oauth_protected_resource_metadata_url(
+                self.settings.auth.resource_server_url
+            )
 
             middleware = [
                 Middleware(
@@ -822,7 +838,11 @@ async def handle_streamable_http(
             routes.append(
                 Mount(
                     self.settings.streamable_http_path,
-                    app=RequireAuthMiddleware(handle_streamable_http, required_scopes),
+                    app=RequireAuthMiddleware(
+                        handle_streamable_http,
+                        required_scopes,
+                        str(resource_metadata_url),
+                    ),
                 )
             )
         else:

tests/client/test_auth.py

@@ -96,7 +96,7 @@ def oauth_token():
 
 
 @pytest.fixture
-async def oauth_provider(client_metadata, mock_storage):
+def oauth_provider(client_metadata, mock_storage):
     async def mock_redirect_handler(url: str) -> None:
         pass
 

tests/server/auth/middleware/test_bearer_auth.py

@@ -307,6 +307,34 @@ async def send(message: Message) -> None:
         assert excinfo.value.detail == "Unauthorized"
         assert not app.called
 
+    async def test_no_user_with_adds_www_authenticate_header(
+        self,
+    ):
+        """Test middleware with no user in scope."""
+        app = MockApp()
+        middleware = RequireAuthMiddleware(
+            app,
+            required_scopes=["read"],
+            resource_metadata_url="https://example.com/.well-known/oauth-protected-resource",
+        )
+        scope: Scope = {"type": "http"}
+
+        async def receive() -> Message:
+            return {"type": "http.request"}
+
+        async def send(message: Message) -> None:
+            pass
+
+        with pytest.raises(HTTPException) as excinfo:
+            await middleware(scope, receive, send)
+
+        assert excinfo.value.status_code == 401
+        assert excinfo.value.detail == "Unauthorized"
+        assert excinfo.value.headers == {
+            "WWW-Authenticate": 'Bearer resource="https://example.com/.well-known/oauth-protected-resource"'
+        }
+        assert not app.called
+
     async def test_non_authenticated_user(self):
         """Test middleware with non-authenticated user in scope."""
         app = MockApp()

tests/server/fastmcp/auth/test_auth_integration.py

@@ -205,15 +205,17 @@ def mock_oauth_provider():
 def auth_app(mock_oauth_provider):
     # Create auth router
     auth_routes = create_auth_routes(
-        mock_oauth_provider,
-        AnyHttpUrl("https://auth.example.com"),
-        AnyHttpUrl("https://docs.example.com"),
+        provider=mock_oauth_provider,
+        issuer_url=AnyHttpUrl("https://auth.example.com"),
+        resource_server_url=AnyHttpUrl("https://api.example.com"),
+        service_documentation_url=AnyHttpUrl("https://docs.example.com"),
         client_registration_options=ClientRegistrationOptions(
             enabled=True,
             valid_scopes=["read", "write", "profile"],
             default_scopes=["read", "write"],
         ),
         revocation_options=RevocationOptions(enabled=True),
+        resource_name="Test Resource Server",
     )
 
     # Create Starlette app
@@ -345,7 +347,27 @@ async def tokens(test_client, registered_client, auth_code, pkce_challenge, requ
 
 class TestAuthEndpoints:
     @pytest.mark.anyio
-    async def test_metadata_endpoint(self, test_client: httpx.AsyncClient):
+    async def test_protected_resource_metadata_endpoint(
+        self, test_client: httpx.AsyncClient
+    ):
+        """Test the OAuth 2.0 protected resource metadata endpoint."""
+        response = await test_client.get("/.well-known/oauth-protected-resource")
+        print(f"Got response: {response.status_code}")
+        if response.status_code != 200:
+            print(f"Response content: {response.content}")
+        assert response.status_code == 200
+
+        metadata = response.json()
+        assert metadata["resource"] == "https://api.example.com/"
+        assert metadata["authorization_servers"] == ["https://auth.example.com/"]
+        assert metadata["resource_name"] == "Test Resource Server"
+        assert metadata["resource_documentation"] == "https://docs.example.com/"
+        assert metadata["scopes_supported"] == ["read", "write", "profile"]
+
+    @pytest.mark.anyio
+    async def test_authorization_server_metadata_endpoint(
+        self, test_client: httpx.AsyncClient
+    ):
         """Test the OAuth 2.0 metadata endpoint."""
         print("Sending request to metadata endpoint")
         response = await test_client.get("/.well-known/oauth-authorization-server")