Merge pull request #300 from jlowin/auth

Add auth support

.github/workflows/run-tests.yml

@@ -59,4 +59,4 @@ jobs:
           uv pip install pyreadline3
 
       - name: Run tests
-        run: uv run pytest -vv
+        run: uv run --frozen pytest -vv

README.md

@@ -760,7 +760,7 @@ Contributions make the open-source community vibrant! We welcome improvements an
 
 Run the test suite:
 ```bash
-uv run pytest -vv
+uv run --frozen pytest -vv
 ```
 
 #### Formatting & Linting

docs/servers/fastmcp.mdx

@@ -292,4 +292,42 @@ print(mcp.settings.on_duplicate_tools) # Output: "error"
 - **`on_duplicate_resources`**: How to handle duplicate resource registrations
 - **`on_duplicate_prompts`**: How to handle duplicate prompt registrations
 
-All of these can be configured directly as parameters when creating the `FastMCP` instance.
+All of these can be configured directly as parameters when creating the `FastMCP` instance.
+
+## Authentication
+
+<VersionBadge version="2.2.7" />
+
+FastMCP inherits support for OAuth 2.0 authentication from the MCP protocol, allowing servers to protect their tools and resources behind authentication.
+
+### OAuth 2.0 Support
+
+The `mcp.server.auth` module implements an OAuth 2.0 server interface that servers can use by providing an implementation of the `OAuthServerProvider` protocol.
+
+```python
+from fastmcp import FastMCP
+from mcp.server.auth.settings import (
+    RevocationOptions,
+    ClientRegistrationOptions,
+    AuthSettings,
+)
+
+
+# Create a server with authentication
+mcp = FastMCP(
+    name="SecureApp",
+    auth_provider=MyOAuthServerProvider(),
+    auth=AuthSettings(
+        issuer_url="https://myapp.com",
+        revocation_options=RevocationOptions(
+            enabled=True,
+        ),
+        client_registration_options=ClientRegistrationOptions(
+            enabled=True,
+            valid_scopes=["myscope", "myotherscope"],
+            default_scopes=["myscope"],
+        ),
+        required_scopes=["myscope"],
+    ),
+)
+```

justfile

@@ -6,4 +6,4 @@ test: build
 
 # Run pyright on all files
 typecheck:
-    uv run pyright
+    uv run --frozen pyright

src/fastmcp/server/server.py

@@ -15,6 +15,12 @@ from typing import TYPE_CHECKING, Any, Generic, Literal
 import anyio
 import httpx
 import uvicorn
+from mcp.server.auth.middleware.auth_context import AuthContextMiddleware
+from mcp.server.auth.middleware.bearer_auth import (
+    BearerAuthBackend,
+    RequireAuthMiddleware,
+)
+from mcp.server.auth.provider import OAuthAuthorizationServerProvider
 from mcp.server.lowlevel.helper_types import ReadResourceContents
 from mcp.server.lowlevel.server import LifespanResultT
 from mcp.server.lowlevel.server import Server as MCPServer
@@ -36,8 +42,12 @@ from mcp.types import ResourceTemplate as MCPResourceTemplate
 from mcp.types import Tool as MCPTool
 from pydantic.networks import AnyUrl
 from starlette.applications import Starlette
+from starlette.middleware import Middleware
+from starlette.middleware.authentication import AuthenticationMiddleware
 from starlette.requests import Request
+from starlette.responses import Response
 from starlette.routing import Mount, Route
+from starlette.types import Receive, Scope, Send
 
 import fastmcp
 import fastmcp.settings
@@ -184,6 +194,8 @@ class FastMCP(Generic[LifespanResultT]):
         self,
         name: str | None = None,
         instructions: str | None = None,
+        auth_server_provider: OAuthAuthorizationServerProvider[Any, Any, Any]
+        | None = None,
         lifespan: (
             Callable[
                 [FastMCP[LifespanResultT]],
@@ -221,6 +233,15 @@ class FastMCP(Generic[LifespanResultT]):
         self._prompt_manager = PromptManager(
             duplicate_behavior=self.settings.on_duplicate_prompts
         )
+
+        if (self.settings.auth is not None) != (auth_server_provider is not None):
+            # TODO: after we support separate authorization servers (see
+            raise ValueError(
+                "settings.auth must be specified if and only if auth_server_provider "
+                "is specified"
+            )
+        self._auth_server_provider = auth_server_provider
+        self._custom_starlette_routes: list[Route] = []
         self.dependencies = self.settings.dependencies
 
         # Set up MCP protocol handlers
@@ -340,6 +361,50 @@ class FastMCP(Generic[LifespanResultT]):
             self._cache.set("prompts", prompts)
         return prompts
 
+    def custom_route(
+        self,
+        path: str,
+        methods: list[str],
+        name: str | None = None,
+        include_in_schema: bool = True,
+    ):
+        """
+        Decorator to register a custom HTTP route on the FastMCP server.
+
+        Allows adding arbitrary HTTP endpoints outside the standard MCP protocol,
+        which can be useful for OAuth callbacks, health checks, or admin APIs.
+        The handler function must be an async function that accepts a Starlette
+        Request and returns a Response.
+
+        Args:
+            path: URL path for the route (e.g., "/oauth/callback")
+            methods: List of HTTP methods to support (e.g., ["GET", "POST"])
+            name: Optional name for the route (to reference this route with
+                Starlette's reverse URL lookup feature)
+            include_in_schema: Whether to include in OpenAPI schema, defaults to True
+
+        Example:
+            @server.custom_route("/health", methods=["GET"])
+            async def health_check(request: Request) -> Response:
+                return JSONResponse({"status": "ok"})
+        """
+
+        def decorator(
+            func: Callable[[Request], Awaitable[Response]],
+        ) -> Callable[[Request], Awaitable[Response]]:
+            self._custom_starlette_routes.append(
+                Route(
+                    path,
+                    endpoint=func,
+                    methods=methods,
+                    name=name,
+                    include_in_schema=include_in_schema,
+                )
+            )
+            return func
+
+        return decorator
+
     async def _mcp_list_tools(self) -> list[MCPTool]:
         """
         List all available tools, in the format expected by the low-level MCP
@@ -770,26 +835,104 @@ class FastMCP(Generic[LifespanResultT]):
 
     def sse_app(self) -> Starlette:
         """Return an instance of the SSE server app."""
+        from starlette.middleware import Middleware
+        from starlette.routing import Mount, Route
+
+        # Set up auth context and dependencies
+
         sse = SseServerTransport(self.settings.message_path)
 
-        async def handle_sse(request: Request) -> None:
+        async def handle_sse(scope: Scope, receive: Receive, send: Send):
+            # Add client ID from auth context into request context if available
+
             async with sse.connect_sse(
-                request.scope,
-                request.receive,
-                request._send,  # type: ignore[reportPrivateUsage]
+                scope,
+                receive,
+                send,
             ) as streams:
                 await self._mcp_server.run(
                     streams[0],
                     streams[1],
                     self._mcp_server.create_initialization_options(),
                 )
+            return Response()
 
+        # Create routes
+        routes: list[Route | Mount] = []
+        middleware: list[Middleware] = []
+        required_scopes = []
+
+        # 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
+
+            required_scopes = self.settings.auth.required_scopes or []
+
+            middleware = [
+                # extract auth info from request (but do not require it)
+                Middleware(
+                    AuthenticationMiddleware,
+                    backend=BearerAuthBackend(
+                        provider=self._auth_server_provider,
+                    ),
+                ),
+                # Add the auth context middleware to store
+                # authenticated user in a contextvar
+                Middleware(AuthContextMiddleware),
+            ]
+            routes.extend(
+                create_auth_routes(
+                    provider=self._auth_server_provider,
+                    issuer_url=self.settings.auth.issuer_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,
+                )
+            )
+
+        # 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),
+                    methods=["GET"],
+                )
+            )
+            routes.append(
+                Mount(
+                    self.settings.message_path,
+                    app=RequireAuthMiddleware(sse.handle_post_message, required_scopes),
+                )
+            )
+        else:
+            # Auth is disabled, no need for RequireAuthMiddleware
+            # Since handle_sse is an ASGI app, we need to create a compatible endpoint
+            async def sse_endpoint(request: Request) -> None:
+                # Convert the Starlette request to ASGI parameters
+                await handle_sse(request.scope, request.receive, request._send)  # type: ignore[reportPrivateUsage]
+
+            routes.append(
+                Route(
+                    self.settings.sse_path,
+                    endpoint=sse_endpoint,
+                    methods=["GET"],
+                )
+            )
+            routes.append(
+                Mount(
+                    self.settings.message_path,
+                    app=sse.handle_post_message,
+                )
+            )
+        # mount these routes last, so they have the lowest route matching precedence
+        routes.extend(self._custom_starlette_routes)
+
+        # Create Starlette app with routes and middleware
         return Starlette(
-            debug=self.settings.debug,
-            routes=[
-                Route(self.settings.sse_path, endpoint=handle_sse),
-                Mount(self.settings.message_path, app=sse.handle_post_message),
-            ],
+            debug=self.settings.debug, routes=routes, middleware=middleware
         )
 
     def mount(

src/fastmcp/settings.py

@@ -2,6 +2,7 @@ from __future__ import annotations as _annotations
 
 from typing import TYPE_CHECKING, Literal
 
+from mcp.server.auth.settings import AuthSettings
 from pydantic import Field
 from pydantic_settings import BaseSettings, SettingsConfigDict
 
@@ -20,6 +21,8 @@ class Settings(BaseSettings):
         env_prefix="FASTMCP_",
         env_file=".env",
         extra="ignore",
+        env_nested_delimiter="__",
+        nested_model_default_partial_update=True,
     )
 
     test_mode: bool = False
@@ -37,6 +40,8 @@ class ServerSettings(BaseSettings):
         env_prefix="FASTMCP_SERVER_",
         env_file=".env",
         extra="ignore",
+        env_nested_delimiter="__",
+        nested_model_default_partial_update=True,
     )
 
     log_level: LOG_LEVEL = Field(default_factory=lambda: Settings().log_level)
@@ -65,6 +70,8 @@ class ServerSettings(BaseSettings):
     # cache settings (for checking mounted servers)
     cache_expiration_seconds: float = 0
 
+    auth: AuthSettings | None = None
+
 
 class ClientSettings(BaseSettings):
     """FastMCP client settings."""