Add OAuth Protected Resource Metadata support

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/client/auth.py

@@ -12,7 +12,7 @@
 import time
 from collections.abc import AsyncGenerator, Awaitable, Callable
 from typing import Protocol
-from urllib.parse import urlencode, urljoin
+from urllib.parse import urlencode, urljoin, urlparse, urlunparse
 
 import anyio
 import httpx
@@ -21,6 +21,7 @@
     OAuthClientInformationFull,
     OAuthClientMetadata,
     OAuthMetadata,
+    OAuthProtectedResourceMetadata,
     OAuthToken,
 )
 from mcp.types import LATEST_PROTOCOL_VERSION
@@ -116,19 +117,59 @@ def _get_authorization_base_url(self, server_url: str) -> str:
 
         Per MCP spec 2.3.2: https://api.example.com/v1/mcp -> https://api.example.com
         """
-        from urllib.parse import urlparse, urlunparse
-
         parsed = urlparse(server_url)
-        # Remove path component
         return urlunparse((parsed.scheme, parsed.netloc, "", "", "", ""))
 
+    async def _discover_protected_resource_metadata(
+        self, resource_server_url: str
+    ) -> OAuthProtectedResourceMetadata | None:
+        """
+        Looks up RFC 9728 OAuth 2.0 Protected Resource Metadata.
+
+        If the server returns a 404 for the well-known endpoint, returns None.
+        """
+        async with httpx.AsyncClient() as client:
+            response = await client.get(
+                urljoin(resource_server_url, "/.well-known/oauth-protected-resource")
+            )
+            if response.status_code == 404:
+                return None
+            response.raise_for_status()
+            metadata_json = response.json()
+            logger.debug(
+                f"OAuth protected resource metadata discovered: {metadata_json}"
+            )
+            return OAuthProtectedResourceMetadata.model_validate(metadata_json)
+
     async def _discover_oauth_metadata(self, server_url: str) -> OAuthMetadata | None:
         """
         Discover OAuth metadata from server's well-known endpoint.
+
+        First tries to discover protected resource metadata and use its authorization
+        server URL if available, otherwise falls back to the server's own well-known.
         """
-        # Extract base URL per MCP spec
-        auth_base_url = self._get_authorization_base_url(server_url)
-        url = urljoin(auth_base_url, "/.well-known/oauth-authorization-server")
+        auth_server_url = self._get_authorization_base_url(server_url)
+
+        try:
+            protected_resource_metadata = (
+                await self._discover_protected_resource_metadata(server_url)
+            )
+
+            if (
+                protected_resource_metadata
+                and protected_resource_metadata.authorization_servers
+                and len(protected_resource_metadata.authorization_servers) > 0
+            ):
+                auth_server_url = str(
+                    protected_resource_metadata.authorization_servers[0]
+                )
+        except Exception as e:
+            logger.warning(
+                "Could not load OAuth Protected Resource metadata, "
+                f"falling back to /.well-known/oauth-authorization-server: {e}"
+            )
+
+        url = urljoin(auth_server_url, "/.well-known/oauth-authorization-server")
         headers = {"MCP-Protocol-Version": LATEST_PROTOCOL_VERSION}
 
         async with httpx.AsyncClient() as client:

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

@@ -4,12 +4,12 @@
 from starlette.responses import Response
 
 from mcp.server.auth.json_response import PydanticJSONResponse
-from mcp.shared.auth import OAuthMetadata
+from mcp.shared.auth import OAuthMetadata, OAuthProtectedResourceMetadata
 
 
 @dataclass
 class MetadataHandler:
-    metadata: OAuthMetadata
+    metadata: OAuthMetadata | OAuthProtectedResourceMetadata
 
     async def handle(self, request: Request) -> Response:
         return PydanticJSONResponse(

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

@@ -67,25 +67,36 @@ class RequireAuthMiddleware:
     auth info in the request state.
     """
 
-    def __init__(self, app: Any, required_scopes: list[str]):
+    def __init__(
+        self,
+        app: Any,
+        required_scopes: list[str] | None = None,
+        resource_metadata_url: str | None = None,
+    ):
         """
         Initialize the middleware.
 
         Args:
             app: ASGI application
-            provider: Authentication provider to validate tokens
             required_scopes: Optional list of scopes that the token must have
+            resource_metadata_url: Optional resource metadata URL
         """
         self.app = app
         self.required_scopes = required_scopes
+        self.resource_metadata_url = resource_metadata_url
 
     async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
         auth_user = scope.get("user")
         if not isinstance(auth_user, AuthenticatedUser):
-            raise HTTPException(status_code=401, detail="Unauthorized")
+            headers = (
+                {"WWW-Authenticate": f'Bearer resource="{self.resource_metadata_url}"'}
+                if self.resource_metadata_url
+                else None
+            )
+            raise HTTPException(status_code=401, detail="Unauthorized", headers=headers)
         auth_credentials = scope.get("auth")
 
-        for required_scope in self.required_scopes:
+        for required_scope in self.required_scopes or []:
             # auth_credentials should always be provided; this is just paranoia
             if (
                 auth_credentials is None

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
@@ -16,7 +17,7 @@
 from mcp.server.auth.middleware.client_auth import ClientAuthenticator
 from mcp.server.auth.provider import OAuthAuthorizationServerProvider
 from mcp.server.auth.settings import ClientRegistrationOptions, RevocationOptions
-from mcp.shared.auth import OAuthMetadata
+from mcp.shared.auth import OAuthMetadata, OAuthProtectedResourceMetadata
 
 
 def validate_issuer_url(url: AnyHttpUrl):
@@ -67,9 +68,11 @@ def cors_middleware(
 def create_auth_routes(
     provider: OAuthAuthorizationServerProvider[Any, Any, Any],
     issuer_url: AnyHttpUrl,
+    resource_server_url: AnyHttpUrl,
     service_documentation_url: AnyHttpUrl | None = None,
     client_registration_options: ClientRegistrationOptions | None = None,
     revocation_options: RevocationOptions | None = None,
+    resource_name: str | None = None,
 ) -> list[Route]:
     validate_issuer_url(issuer_url)
 
@@ -85,11 +88,27 @@ def create_auth_routes(
     )
     client_authenticator = ClientAuthenticator(provider)
 
+    protected_resource_metadata = OAuthProtectedResourceMetadata(
+        resource=resource_server_url,
+        authorization_servers=[metadata.issuer],
+        scopes_supported=metadata.scopes_supported,
+        resource_name=resource_name,
+        resource_documentation=service_documentation_url,
+    )
+
     # Create routes
     # Allow CORS requests for endpoints meant to be hit by the OAuth client
     # (with the client secret). This is intended to support things like MCP Inspector,
     # where the client runs in a web browser.
     routes = [
+        Route(
+            "/.well-known/oauth-protected-resource",
+            endpoint=cors_middleware(
+                MetadataHandler(protected_resource_metadata).handle,
+                ["GET", "OPTIONS"],
+            ),
+            methods=["GET", "OPTIONS"],
+        ),
         Route(
             "/.well-known/oauth-authorization-server",
             endpoint=cors_middleware(
@@ -189,3 +208,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/auth/settings.py

@@ -15,10 +15,15 @@ class RevocationOptions(BaseModel):
 class AuthSettings(BaseModel):
     issuer_url: AnyHttpUrl = Field(
         ...,
-        description="URL advertised as OAuth issuer; this should be the URL the server "
-        "is reachable at",
+        description="The authorization server's issuer identifier",
+    )
+    resource_server_url: AnyHttpUrl = Field(
+        ..., description="URL of the MCP server, for use in protected resource metadata"
     )
     service_documentation_url: AnyHttpUrl | None = None
     client_registration_options: ClientRegistrationOptions | None = None
     revocation_options: RevocationOptions | None = None
     required_scopes: list[str] | None = None
+    resource_name: str | None = Field(
+        None, description="Optional resource name to display in resource metadata"
+    )

src/mcp/server/fastmcp/server.py

@@ -697,9 +697,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)
@@ -717,26 +723,32 @@ async def handle_sse(scope: Scope, receive: Receive, send: Send):
                 create_auth_routes(
                     provider=self._auth_server_provider,
                     issuer_url=self.settings.auth.issuer_url,
+                    resource_server_url=self.settings.auth.resource_server_url,
                     service_documentation_url=self.settings.auth.service_documentation_url,
                     client_registration_options=self.settings.auth.client_registration_options,
                     revocation_options=self.settings.auth.revocation_options,
+                    resource_name=self.settings.auth.resource_name,
                 )
             )
 
-        # 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 +807,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(
@@ -812,15 +830,21 @@ async def handle_streamable_http(
                 create_auth_routes(
                     provider=self._auth_server_provider,
                     issuer_url=self.settings.auth.issuer_url,
+                    resource_server_url=self.settings.auth.resource_server_url,
                     service_documentation_url=self.settings.auth.service_documentation_url,
                     client_registration_options=self.settings.auth.client_registration_options,
                     revocation_options=self.settings.auth.revocation_options,
+                    resource_name=self.settings.auth.resource_name,
                 )
             )
             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:

src/mcp/shared/auth.py

@@ -135,3 +135,25 @@ class OAuthMetadata(BaseModel):
     ) = None
     introspection_endpoint_auth_signing_alg_values_supported: None = None
     code_challenge_methods_supported: list[Literal["S256"]] | None = None
+
+
+class OAuthProtectedResourceMetadata(BaseModel):
+    """
+    RFC 9728 OAuth Protected Resource Metadata
+    See https://datatracker.ietf.org/doc/html/rfc9728
+    """
+
+    resource: AnyHttpUrl
+    authorization_servers: list[AnyHttpUrl] | None = None
+    jwks_uri: AnyHttpUrl | None = None
+    scopes_supported: list[str] | None = None
+    bearer_methods_supported: list[str] | None = None
+    resource_signing_alg_values_supported: list[str] | None = None
+    resource_name: str | None = None
+    resource_documentation: AnyHttpUrl | None = None
+    resource_policy_uri: AnyHttpUrl | None = None
+    resource_tos_uri: AnyHttpUrl | None = None
+    tls_client_certificate_bound_access_tokens: bool | None = None
+    authorization_details_types_supported: list[str] | None = None
+    dpop_signing_alg_values_supported: list[str] | None = None
+    dpop_bound_access_tokens_required: bool | None = None