refactor: extract auth components and streamable HTTP app helpers

- Add build_auth_components() in mcp.server.auth.components for reusable
  auth setup (middleware, endpoint wrapper, routes)
- Refactor create_streamable_http_app() to take session_manager as first
  arg with keyword args for app config (removed StreamableHTTPAppConfig)
- FastMCP now uses _build_auth_components() helper, reducing duplication
  between sse_app() and streamable_http_app()
- Session manager is now created/owned by caller, passed to app creator
- Add unit tests for build_auth_components()
- Export AuthComponents, build_auth_components from mcp.server.auth
- Export StreamableHTTPSessionManager, create_streamable_http_app from
  mcp.server

Claude-Generated-By: Claude Code (cli/claude-opus-4-5=100%)
Claude-Steers: 14
Claude-Permission-Prompts: 13
Claude-Escapes: 0
Claude-Plan:
<claude-plan>
# Plan: Extract Auth Helper and Make FastMCP a Thin Wrapper

## Summary

Create a shared `build_auth_components()` helper in the auth module that both `sse_app()` and `streamable_http_app()` can use. This removes ~60 lines of duplicated auth logic from FastMCP and makes it a much thinner wrapper.

## Files to Modify

1. **`src/mcp/server/auth/routes.py`** - Add `AuthConfig` dataclass and `build_auth_components()` function
2. **`src/mcp/server/fastmcp/server.py`** - Refactor both `sse_app()` and `streamable_http_app()` to use the helper
3. **`src/mcp/server/__init__.py`** - Export new auth helper for low-level users

## Implementation

### Step 1: Add to `src/mcp/server/auth/routes.py`

**Add new dataclass and helper function:**

```python
from dataclasses import dataclass, field
from starlette.middleware import Middleware
from starlette.middleware.authentication import AuthenticationMiddleware
from starlette.types import ASGIApp

@DataClass
class AuthConfig:
    """Configuration for auth components in Starlette apps."""

    # Token verification (required)
    token_verifier: TokenVerifier

    # Auth settings
    issuer_url: AnyHttpUrl
    required_scopes: list[str] = field(default_factory=list)
    resource_server_url: AnyHttpUrl | None = None

    # Optional: Full OAuth AS provider for serving auth endpoints
    auth_server_provider: OAuthAuthorizationServerProvider[Any, Any, Any] | None = None
    service_documentation_url: AnyHttpUrl | None = None
    client_registration_options: ClientRegistrationOptions | None = None
    revocation_options: RevocationOptions | None = None

@DataClass
class AuthComponents:
    """Auth components ready to be used in a Starlette app."""

    routes: list[Route]
    middleware: list[Middleware]
    endpoint_wrapper: Callable[[ASGIApp], ASGIApp]

def build_auth_components(config: AuthConfig) -> AuthComponents:
    """
    Build auth routes, middleware, and endpoint wrapper from config.

    Returns an AuthComponents with:
    - routes: OAuth AS routes (if provider set) + protected resource metadata
    - middleware: AuthenticationMiddleware + AuthContextMiddleware
    - endpoint_wrapper: RequireAuthMiddleware wrapper function
    """
    routes: list[Route] = []

    # Build middleware
    middleware = [
        Middleware(
            AuthenticationMiddleware,
            backend=BearerAuthBackend(config.token_verifier),
        ),
        Middleware(AuthContextMiddleware),
    ]

    # Add OAuth AS routes if provider is configured
    if config.auth_server_provider:
        routes.extend(
            create_auth_routes(
                provider=config.auth_server_provider,
                issuer_url=config.issuer_url,
                service_documentation_url=config.service_documentation_url,
                client_registration_options=config.client_registration_options,
                revocation_options=config.revocation_options,
            )
        )

    # Add protected resource metadata routes if resource_server_url is set
    if config.resource_server_url:
        routes.extend(
            create_protected_resource_routes(
                resource_url=config.resource_server_url,
                authorization_servers=[config.issuer_url],
                scopes_supported=config.required_scopes or None,
            )
        )

    # Build endpoint wrapper
    resource_metadata_url = None
    if config.resource_server_url:
        resource_metadata_url = build_resource_metadata_url(config.resource_server_url)

    def endpoint_wrapper(app: ASGIApp) -> ASGIApp:
        return RequireAuthMiddleware(app, config.required_scopes, resource_metadata_url)

    return AuthComponents(
        routes=routes,
        middleware=middleware,
        endpoint_wrapper=endpoint_wrapper,
    )
```

### Step 2: Refactor `FastMCP.streamable_http_app()`

Replace the ~50 lines of auth logic with:

```python
def streamable_http_app(self) -> Starlette:
    """Return an instance of the StreamableHTTP server app."""
    additional_routes: list[Route | Mount] = []
    middleware: list[Middleware] = []
    endpoint_wrapper: Callable[[ASGIApp], ASGIApp] | None = None

    # Build auth components if auth is configured
    if self.settings.auth and self._token_verifier:
        from mcp.server.auth.routes import AuthConfig, build_auth_components

        auth_config = AuthConfig(
            token_verifier=self._token_verifier,
            issuer_url=self.settings.auth.issuer_url,
            required_scopes=self.settings.auth.required_scopes or [],
            resource_server_url=self.settings.auth.resource_server_url,
            auth_server_provider=self._auth_server_provider,
            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,
        )
        auth_components = build_auth_components(auth_config)

        additional_routes.extend(auth_components.routes)
        middleware = auth_components.middleware
        endpoint_wrapper = auth_components.endpoint_wrapper

    # Add custom routes last
    additional_routes.extend(self._custom_starlette_routes)

    # Create config and call low-level function
    config = StreamableHTTPAppConfig(
        mcp_server=self._mcp_server,
        event_store=self._event_store,
        retry_interval=self._retry_interval,
        json_response=self.settings.json_response,
        stateless=self.settings.stateless_http,
        security_settings=self.settings.transport_security,
        endpoint_path=self.settings.streamable_http_path,
        debug=self.settings.debug,
        additional_routes=additional_routes,
        middleware=middleware,
        endpoint_wrapper=endpoint_wrapper,
    )

    starlette_app, session_manager = create_streamable_http_app(config)
    self._session_manager = session_manager
    return starlette_app
```

### Step 3: Refactor `FastMCP.sse_app()`

Similar refactor - replace the auth logic with `build_auth_components()`. The SSE app has slightly different route structure (two endpoints: SSE and messages), so the wrapper is applied to each endpoint individually rather than via config:

```python
def sse_app(self, mount_path: str | None = None) -> Starlette:
    """Return an instance of the SSE server app."""
    if mount_path is not None:
        self.settings.mount_path = mount_path

    normalized_message_endpoint = self._normalize_path(
        self.settings.mount_path, self.settings.message_path
    )

    sse = SseServerTransport(
        normalized_message_endpoint,
        security_settings=self.settings.transport_security,
    )

    async def handle_sse(scope: Scope, receive: Receive, send: Send):
        async with sse.connect_sse(scope, receive, send) as streams:
            await self._mcp_server.run(
                streams[0], streams[1],
                self._mcp_server.create_initialization_options(),
            )
        return Response()

    routes: list[Route | Mount] = []
    middleware: list[Middleware] = []

    # Build auth components if configured
    if self.settings.auth and self._token_verifier:
        from mcp.server.auth.routes import AuthConfig, build_auth_components

        auth_config = AuthConfig(
            token_verifier=self._token_verifier,
            issuer_url=self.settings.auth.issuer_url,
            required_scopes=self.settings.auth.required_scopes or [],
            resource_server_url=self.settings.auth.resource_server_url,
            auth_server_provider=self._auth_server_provider,
            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,
        )
        auth_components = build_auth_components(auth_config)

        routes.extend(auth_components.routes)
        middleware = auth_components.middleware

        # SSE has two endpoints that need wrapping
        routes.append(Route(
            self.settings.sse_path,
            endpoint=auth_components.endpoint_wrapper(handle_sse),
            methods=["GET"],
        ))
        routes.append(Mount(
            self.settings.message_path,
            app=auth_components.endpoint_wrapper(sse.handle_post_message),
        ))
    else:
        # No auth - add routes directly
        async def sse_endpoint(request: Request) -> Response:
            return await handle_sse(request.scope, request.receive, request._send)

        routes.append(Route(self.settings.sse_path, endpoint=sse_endpoint, methods=["GET"]))
        routes.append(Mount(self.settings.message_path, app=sse.handle_post_message))

    routes.extend(self._custom_starlette_routes)
    return Starlette(debug=self.settings.debug, routes=routes, middleware=middleware)
```

### Step 4: Update exports in `src/mcp/server/__init__.py`

Add the new auth helper to exports:

```python
from .auth.routes import AuthConfig, AuthComponents, build_auth_components
```

## Verification

1. Run existing tests:
   ```bash
   PYTEST_DISABLE_PLUGIN_AUTOLOAD="" uv run --frozen pytest tests/server/fastmcp/
   ```

2. Run auth tests specifically:
   ```bash
   PYTEST_DISABLE_PLUGIN_AUTOLOAD="" uv run --frozen pytest tests/server/fastmcp/auth/
   ```

3. Run type checking:
   ```bash
   uv run --frozen pyright
   ```

4. Test low-level usage with auth:
   ```python
   from mcp.server import Server, StreamableHTTPAppConfig, create_streamable_http_app
   from mcp.server.auth.routes import AuthConfig, build_auth_components

   server = Server('test')
   auth = build_auth_components(AuthConfig(...))

   config = StreamableHTTPAppConfig(
       mcp_server=server,
       additional_routes=auth.routes,
       middleware=auth.middleware,
       endpoint_wrapper=auth.endpoint_wrapper,
   )
   app, manager = create_streamable_http_app(config)
   ```

## Result

FastMCP's `streamable_http_app()` goes from ~90 lines to ~30 lines, and `sse_app()` similarly shrinks. The auth logic is now:
- Reusable by low-level Server users
- Testable in isolation
- Shared between SSE and StreamableHTTP transports
</claude-plan>

Author: maxisbey

src/mcp/server/__init__.py

@@ -1,5 +1,18 @@
 from .fastmcp import FastMCP
 from .lowlevel import NotificationOptions, Server
 from .models import InitializationOptions
+from .streamable_http_manager import (
+    StreamableHTTPASGIApp,
+    StreamableHTTPSessionManager,
+    create_streamable_http_app,
+)
 
-__all__ = ["Server", "FastMCP", "NotificationOptions", "InitializationOptions"]
+__all__ = [
+    "Server",
+    "FastMCP",
+    "NotificationOptions",
+    "InitializationOptions",
+    "StreamableHTTPASGIApp",
+    "StreamableHTTPSessionManager",
+    "create_streamable_http_app",
+]

src/mcp/server/auth/__init__.py

@@ -1,3 +1,10 @@
 """
 MCP OAuth server authorization components.
 """
+
+from mcp.server.auth.components import AuthComponents, build_auth_components
+
+__all__ = [
+    "AuthComponents",
+    "build_auth_components",
+]

src/mcp/server/auth/components.py

@@ -0,0 +1,132 @@
+"""Auth components for MCP servers."""
+
+from collections.abc import Callable, Sequence
+from dataclasses import dataclass
+from typing import Any
+
+from pydantic import AnyHttpUrl
+from starlette.middleware import Middleware
+from starlette.middleware.authentication import AuthenticationMiddleware
+from starlette.routing import Route
+from starlette.types import ASGIApp
+
+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, TokenVerifier
+from mcp.server.auth.routes import (
+    build_resource_metadata_url,
+    create_auth_routes,
+    create_protected_resource_routes,
+)
+from mcp.server.auth.settings import ClientRegistrationOptions, RevocationOptions
+
+
+@dataclass
+class AuthComponents:
+    """Auth components ready to be used in a Starlette app.
+
+    Attributes:
+        middleware: Authentication middleware to add to the app.
+        endpoint_wrapper: Wrapper function to protect endpoints with auth.
+        routes: OAuth and/or protected resource metadata routes.
+    """
+
+    middleware: list[Middleware]
+    endpoint_wrapper: Callable[[Any], ASGIApp]
+    routes: list[Route]
+
+
+def build_auth_components(
+    token_verifier: TokenVerifier,
+    *,
+    required_scopes: Sequence[str] = (),
+    # OAuth AS routes (only if MCP server IS the auth server)
+    auth_server_provider: OAuthAuthorizationServerProvider[Any, Any, Any] | None = None,
+    issuer_url: AnyHttpUrl | None = None,
+    service_documentation_url: AnyHttpUrl | None = None,
+    client_registration_options: ClientRegistrationOptions | None = None,
+    revocation_options: RevocationOptions | None = None,
+    # Protected resource metadata routes
+    resource_server_url: AnyHttpUrl | None = None,
+) -> AuthComponents:
+    """
+    Build auth components for a Starlette app.
+
+    This function creates the middleware, endpoint wrapper, and routes needed
+    to add OAuth 2.0 authentication to an MCP server.
+
+    Args:
+        token_verifier: Verifies bearer tokens from requests.
+        required_scopes: Scopes required to access the MCP endpoint.
+        auth_server_provider: OAuth AS provider (if MCP server is the auth server).
+        issuer_url: OAuth issuer URL (required if auth_server_provider is set).
+        service_documentation_url: URL to service documentation.
+        client_registration_options: Options for dynamic client registration.
+        revocation_options: Options for token revocation.
+        resource_server_url: Resource server URL for protected resource metadata.
+
+    Returns:
+        AuthComponents containing middleware, endpoint wrapper, and routes.
+
+    Example:
+        >>> auth = build_auth_components(
+        ...     token_verifier=my_verifier,
+        ...     required_scopes=["mcp:read"],
+        ...     resource_server_url="https://mcp.example.com",
+        ... )
+        >>> app = create_streamable_http_app(
+        ...     session_manager,
+        ...     middleware=auth.middleware,
+        ...     endpoint_wrapper=auth.endpoint_wrapper,
+        ...     additional_routes=auth.routes,
+        ... )
+    """
+    routes: list[Route] = []
+
+    # Build middleware
+    middleware = [
+        Middleware(
+            AuthenticationMiddleware,
+            backend=BearerAuthBackend(token_verifier),
+        ),
+        Middleware(AuthContextMiddleware),
+    ]
+
+    # Add OAuth AS routes if provider is configured
+    if auth_server_provider is not None:
+        if issuer_url is None:
+            raise ValueError("issuer_url is required when auth_server_provider is set")
+        routes.extend(
+            create_auth_routes(
+                provider=auth_server_provider,
+                issuer_url=issuer_url,
+                service_documentation_url=service_documentation_url,
+                client_registration_options=client_registration_options,
+                revocation_options=revocation_options,
+            )
+        )
+
+    # Add protected resource metadata routes if resource_server_url is set
+    if resource_server_url is not None:
+        authorization_servers = [issuer_url] if issuer_url is not None else []
+        routes.extend(
+            create_protected_resource_routes(
+                resource_url=resource_server_url,
+                authorization_servers=authorization_servers,
+                scopes_supported=list(required_scopes) if required_scopes else None,
+            )
+        )
+
+    # Build endpoint wrapper
+    resource_metadata_url = None
+    if resource_server_url is not None:
+        resource_metadata_url = build_resource_metadata_url(resource_server_url)
+
+    def endpoint_wrapper(app: Any) -> ASGIApp:
+        return RequireAuthMiddleware(app, list(required_scopes), resource_metadata_url)
+
+    return AuthComponents(
+        middleware=middleware,
+        endpoint_wrapper=endpoint_wrapper,
+        routes=routes,
+    )