Skip to content

Conversation

@maxisbey
Copy link
Contributor

Summary

Refactors the auth and streamable HTTP setup to be more composable and reusable by low-level Server users.

Changes

New build_auth_components() function

  • Lives in mcp.server.auth.components
  • Takes token_verifier and optional config, returns AuthComponents with:
    • middleware - Authentication middleware
    • endpoint_wrapper - Function to protect endpoints
    • routes - OAuth AS and/or protected resource metadata routes
  • Reusable by both FastMCP and low-level Server users

Refactored create_streamable_http_app()

  • Now takes session_manager as first argument (caller owns it)
  • Keyword args for app configuration (removed StreamableHTTPAppConfig dataclass)
  • Returns just Starlette, not a tuple
  • Clear ownership: caller creates session manager, helper creates app

FastMCP improvements

  • Added _build_auth_components() private helper
  • Both sse_app() and streamable_http_app() use shared auth setup
  • Reduced code duplication

New exports

  • mcp.server.auth: AuthComponents, build_auth_components
  • mcp.server: StreamableHTTPSessionManager, create_streamable_http_app

Example: Low-level usage with auth

from mcp.server import Server, StreamableHTTPSessionManager, create_streamable_http_app
from mcp.server.auth import build_auth_components

server = Server("my-server")
session_manager = StreamableHTTPSessionManager(app=server, ...)

auth = build_auth_components(
    token_verifier=my_verifier,
    issuer_url="https://auth.example.com",
    required_scopes=["mcp:read"],
)

app = create_streamable_http_app(
    session_manager,
    additional_routes=auth.routes,
    middleware=auth.middleware,
    endpoint_wrapper=auth.endpoint_wrapper,
)

Testing

  • Added unit tests for build_auth_components() in tests/server/auth/test_components.py
  • All existing tests pass

- 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>
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

None yet

Projects

None yet

Development

Successfully merging this pull request may close these issues.

2 participants