server

Author: ihrpr

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

@@ -325,8 +325,7 @@ def create_authorization_server(settings: AuthServerSettings) -> Starlette:
             default_scopes=[settings.mcp_scope],
         ),
         required_scopes=[settings.mcp_scope],
-        resource_url=settings.server_url,
-        resource_name="MCP Authorization Server",
+        authorization_servers=None,
     )
 
     # Create OAuth routes
@@ -336,8 +335,6 @@ def create_authorization_server(settings: AuthServerSettings) -> Starlette:
         service_documentation_url=auth_settings.service_documentation_url,
         client_registration_options=auth_settings.client_registration_options,
         revocation_options=auth_settings.revocation_options,
-        resource_url=settings.server_url,  # Enable protected resource metadata
-        resource_name="MCP Authorization Server",
     )
 
     # Add GitHub callback route

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

@@ -15,15 +15,11 @@
 import httpx
 from pydantic import AnyHttpUrl
 from pydantic_settings import BaseSettings, SettingsConfigDict
-from starlette.authentication import AuthCredentials, AuthenticationBackend
-from starlette.requests import HTTPConnection
-from starlette.responses import JSONResponse
 
 from mcp.server.auth.middleware.auth_context import get_access_token
-from mcp.server.auth.middleware.bearer_auth import AuthenticatedUser
-from mcp.server.auth.provider import AccessToken
+from mcp.server.auth.settings import AuthSettings
+from mcp.server.auth.token_verifier import IntrospectionTokenVerifier
 from mcp.server.fastmcp.server import FastMCP
-from mcp.shared.auth import ProtectedResourceMetadata
 
 logger = logging.getLogger(__name__)
 
@@ -51,63 +47,6 @@ def __init__(self, **data):
         super().__init__(**data)
 
 
-class TokenIntrospectionAuthBackend(AuthenticationBackend):
-    """
-    Authentication backend for Resource Server that validates tokens via AS introspection.
-
-    This backend:
-    1. Extracts Bearer tokens from Authorization header
-    2. Calls Authorization Server's introspection endpoint
-    3. Creates AuthenticatedUser from token info
-    """
-
-    def __init__(self, settings: ResourceServerSettings):
-        self.settings = settings
-        self.introspection_endpoint = settings.auth_server_introspection_endpoint
-
-    async def authenticate(self, conn: HTTPConnection):
-        auth_header = next(
-            (conn.headers.get(key) for key in conn.headers if key.lower() == "authorization"),
-            None,
-        )
-        if not auth_header or not auth_header.lower().startswith("bearer "):
-            return None
-
-        token = auth_header[7:]  # Remove "Bearer " prefix
-
-        # Introspect token with Authorization Server
-        async with httpx.AsyncClient() as client:
-            try:
-                response = await client.post(
-                    self.introspection_endpoint,
-                    data={"token": token},
-                    headers={"Content-Type": "application/x-www-form-urlencoded"},
-                )
-
-                if response.status_code != 200:
-                    logger.debug(f"Token introspection failed with status {response.status_code}")
-                    return None
-
-                data = response.json()
-                if not data.get("active", False):
-                    logger.debug("Token is not active")
-                    return None
-
-                # Create auth info from introspection response
-                auth_info = AccessToken(
-                    token=token,
-                    client_id=data.get("client_id", "unknown"),
-                    scopes=data.get("scope", "").split() if data.get("scope") else [],
-                    expires_at=data.get("exp"),
-                )
-
-                return AuthCredentials(auth_info.scopes), AuthenticatedUser(auth_info)
-
-            except Exception:
-                logger.exception("Token introspection failed")
-                return None
-
-
 def create_resource_server(settings: ResourceServerSettings) -> FastMCP:
     """
     Create MCP Resource Server with token introspection.
@@ -117,35 +56,24 @@ def create_resource_server(settings: ResourceServerSettings) -> FastMCP:
     2. Validates tokens via Authorization Server introspection
     3. Serves MCP tools and resources
     """
-    # Create FastMCP server WITHOUT auth settings (since we'll use custom middleware)
-    # This avoids the FastMCP validation error that requires auth_server_provider
+    # Create token verifier for introspection
+    token_verifier = IntrospectionTokenVerifier(settings.auth_server_introspection_endpoint)
+
+    # Create FastMCP server as a Resource Server
     app = FastMCP(
         name="MCP Resource Server",
         instructions="Resource Server that validates tokens via Authorization Server introspection",
         host=settings.host,
         port=settings.port,
         debug=True,
-        # No auth settings - this is RS, not AS
-    )
-
-    # Add the protected resource metadata route using FastMCP's custom_route
-    @app.custom_route("/.well-known/oauth-protected-resource", methods=["GET", "OPTIONS"])
-    async def protected_resource_metadata(_request):
-        """Handle requests for protected resource metadata."""
-        metadata = ProtectedResourceMetadata(
-            resource=settings.server_url,
+        # Auth configuration for RS mode
+        token_verifier=token_verifier,
+        auth=AuthSettings(
+            issuer_url=settings.server_url,
+            required_scopes=[settings.mcp_scope],
             authorization_servers=[settings.auth_server_url],
-            scopes_supported=[settings.mcp_scope],
-            bearer_methods_supported=["header"],
-        )
-        # Convert to dict with string URLs for JSON serialization
-        response_data = {
-            "resource": str(metadata.resource),
-            "authorization_servers": [str(url) for url in metadata.authorization_servers],
-            "scopes_supported": metadata.scopes_supported,
-            "bearer_methods_supported": metadata.bearer_methods_supported,
-        }
-        return JSONResponse(response_data)
+        ),
+    )
 
     async def get_github_user_data() -> dict[str, Any]:
         """

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

@@ -1,13 +1,14 @@
+import json
 import time
 from typing import Any
 
 from pydantic import AnyHttpUrl
 from starlette.authentication import AuthCredentials, AuthenticationBackend, SimpleUser
-from starlette.exceptions import HTTPException
 from starlette.requests import HTTPConnection
 from starlette.types import Receive, Scope, Send
 
-from mcp.server.auth.provider import AccessToken, OAuthAuthorizationServerProvider
+from mcp.server.auth.provider import AccessToken
+from mcp.server.auth.token_verifier import TokenVerifier
 
 
 class AuthenticatedUser(SimpleUser):
@@ -21,14 +22,11 @@ def __init__(self, auth_info: AccessToken):
 
 class BearerAuthBackend(AuthenticationBackend):
     """
-    Authentication backend that validates Bearer tokens.
+    Authentication backend that validates Bearer tokens using a TokenVerifier.
     """
 
-    def __init__(
-        self,
-        provider: OAuthAuthorizationServerProvider[Any, Any, Any],
-    ):
-        self.provider = provider
+    def __init__(self, token_verifier: TokenVerifier):
+        self.token_verifier = token_verifier
 
     async def authenticate(self, conn: HTTPConnection):
         auth_header = next(
@@ -40,8 +38,8 @@ async def authenticate(self, conn: HTTPConnection):
 
         token = auth_header[7:]  # Remove "Bearer " prefix
 
-        # Validate the token with the provider
-        auth_info = await self.provider.load_access_token(token)
+        # Validate the token with the verifier
+        auth_info = await self.token_verifier.verify_token(token)
 
         if not auth_info:
             return None
@@ -65,7 +63,6 @@ def __init__(
         app: Any,
         required_scopes: list[str],
         resource_metadata_url: AnyHttpUrl | None = None,
-        realm: str | None = None,
     ):
         """
         Initialize the middleware.
@@ -74,22 +71,57 @@ def __init__(
             app: ASGI application
             required_scopes: List of scopes that the token must have
             resource_metadata_url: Optional protected resource metadata URL for WWW-Authenticate header
-            realm: Optional realm for WWW-Authenticate header
         """
         self.app = app
         self.required_scopes = required_scopes
         self.resource_metadata_url = resource_metadata_url
-        self.realm = realm or "mcp"
 
     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")
+            await self._send_auth_error(
+                send, status_code=401, error="invalid_token", description="Authentication required"
+            )
+            return
+
         auth_credentials = scope.get("auth")
 
         for required_scope in self.required_scopes:
             # auth_credentials should always be provided; this is just paranoia
             if auth_credentials is None or required_scope not in auth_credentials.scopes:
-                raise HTTPException(status_code=403, detail="Insufficient scope")
+                await self._send_auth_error(
+                    send, status_code=403, error="insufficient_scope", description=f"Required scope: {required_scope}"
+                )
+                return
 
         await self.app(scope, receive, send)
+
+    async def _send_auth_error(self, send: Send, status_code: int, error: str, description: str) -> None:
+        """Send an authentication error response with WWW-Authenticate header."""
+        # Build WWW-Authenticate header value
+        www_auth_parts = [f'error="{error}"', f'error_description="{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
+        await send(
+            {
+                "type": "http.response.start",
+                "status": status_code,
+                "headers": [
+                    (b"content-type", b"application/json"),
+                    (b"www-authenticate", www_authenticate.encode()),
+                ],
+            }
+        )
+
+        # Send body
+        body = {"error": error, "error_description": description}
+        await send(
+            {
+                "type": "http.response.body",
+                "body": json.dumps(body).encode(),
+            }
+        )

src/mcp/server/auth/routes.py

@@ -9,15 +9,15 @@
 from starlette.types import ASGIApp
 
 from mcp.server.auth.handlers.authorize import AuthorizationHandler
-from mcp.server.auth.handlers.metadata import MetadataHandler, ProtectedResourceMetadataHandler
+from mcp.server.auth.handlers.metadata import MetadataHandler
 from mcp.server.auth.handlers.register import RegistrationHandler
 from mcp.server.auth.handlers.revoke import RevocationHandler
 from mcp.server.auth.handlers.token import TokenHandler
 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.server.streamable_http import MCP_PROTOCOL_VERSION_HEADER
-from mcp.shared.auth import OAuthMetadata, ProtectedResourceMetadata
+from mcp.shared.auth import OAuthMetadata
 
 
 def validate_issuer_url(url: AnyHttpUrl):
@@ -67,8 +67,6 @@ def create_auth_routes(
     service_documentation_url: AnyHttpUrl | None = None,
     client_registration_options: ClientRegistrationOptions | None = None,
     revocation_options: RevocationOptions | None = None,
-    resource_url: AnyHttpUrl | None = None,
-    resource_name: str | None = None,
 ) -> list[Route]:
     validate_issuer_url(issuer_url)
 
@@ -97,25 +95,6 @@ def create_auth_routes(
         ),
     ]
 
-    # Add protected resource metadata endpoint if resource is configured
-    if resource_url:
-        protected_resource_metadata = build_protected_resource_metadata(
-            resource_url,
-            issuer_url,
-            client_registration_options,
-            resource_name,
-        )
-        routes.append(
-            Route(
-                "/.well-known/oauth-protected-resource",
-                endpoint=cors_middleware(
-                    ProtectedResourceMetadataHandler(protected_resource_metadata).handle,
-                    ["GET", "OPTIONS"],
-                ),
-                methods=["GET", "OPTIONS"],
-            )
-        )
-
     # Add remaining auth routes
     routes.extend(
         [
@@ -209,34 +188,38 @@ def build_metadata(
     return metadata
 
 
-def build_protected_resource_metadata(
+def create_protected_resource_routes(
     resource_url: AnyHttpUrl,
-    issuer_url: AnyHttpUrl,
-    client_registration_options: ClientRegistrationOptions,
-    resource_name: str | None = None,
-) -> ProtectedResourceMetadata:
+    authorization_servers: list[AnyHttpUrl],
+    scopes_supported: list[str] | None = None,
+) -> list[Route]:
     """
-    Build protected resource metadata according to RFC 9728.
-
+    Create routes for OAuth 2.0 Protected Resource Metadata (RFC 9728).
+    
     Args:
-        resource_url: The resource server URL
-        issuer_url: The authorization server URL
-        client_registration_options: Client registration options for scopes
-        resource_name: Optional resource name
-
+        resource_url: The URL of this resource server
+        authorization_servers: List of authorization servers that can issue tokens
+        scopes_supported: Optional list of scopes supported by this resource
+        
     Returns:
-        ProtectedResourceMetadata: The protected resource metadata
+        List of Starlette routes for protected resource metadata
     """
+    from mcp.server.auth.handlers.metadata import ProtectedResourceMetadataHandler
+    from mcp.shared.auth import ProtectedResourceMetadata
+    
     metadata = ProtectedResourceMetadata(
         resource=resource_url,
-        authorization_servers=[issuer_url],
-        scopes_supported=client_registration_options.valid_scopes,
-        bearer_methods_supported=["header"],
+        authorization_servers=authorization_servers,
+        scopes_supported=scopes_supported,
+        # bearer_methods_supported defaults to ["header"] in the model
     )
-
-    if resource_name:
-        # Set resource documentation URL if resource name is provided
-        # This could be enhanced to include actual documentation URLs
-        pass
-
-    return metadata
+    
+    handler = ProtectedResourceMetadataHandler(metadata)
+    
+    return [
+        Route(
+            "/.well-known/oauth-protected-resource",
+            endpoint=cors_middleware(handler.handle, ["GET", "OPTIONS"]),
+            methods=["GET", "OPTIONS"],
+        )
+    ]