Keep the public server signatures resolvable by typing.get_type_hints

Moving the HTTP transport and auth imports under TYPE_CHECKING made
typing.get_type_hints() raise NameError on public methods it used to
resolve on: MCPServer.streamable_http_app / sse_app / run_sse_async /
run_streamable_http_async / session_manager and Server.streamable_http_app
/ session_manager (EventStore, TransportSecuritySettings,
StreamableHTTPSessionManager, AuthSettings, OAuthAuthorizationServerProvider,
TokenVerifier were typing-only names in real annotations).

Two of those types get web-framework-free homes so the server modules can
import them for real without loading starlette: the resumability contract
(EventStore, EventMessage, EventCallback, EventId, StreamId) moves to
mcp.server.event_store, and mcp.server.transport_security keeps only the
pydantic TransportSecuritySettings while its starlette middleware moves
beside the transports. Both old import paths keep working with the same
objects. The session manager and the OAuth annotations are spelled through
the lazy mcp.server namespace instead (the trick already used for
Client.server), so they evaluate on demand and MCPServer stops importing
the OAuth provider stack at module load. The starlette-owned annotations
of the app builders (Starlette, Route, Request/Response) stay typing-only;
get_type_hints on those methods needs starlette's names supplied by the
caller.

Author: maxisbey

src/mcp/server/_streamable_http_modern.py

@@ -47,10 +47,11 @@
 from starlette.responses import Response
 from starlette.types import Receive, Scope, Send
 
+from mcp.server._transport_security_middleware import TransportSecurityMiddleware
 from mcp.server.connection import Connection
 from mcp.server.runner import modern_error_data, serve_one
 from mcp.server.streamable_http import check_accept_headers
-from mcp.server.transport_security import TransportSecurityMiddleware, TransportSecuritySettings
+from mcp.server.transport_security import TransportSecuritySettings
 from mcp.shared.dispatcher import CallOptions
 from mcp.shared.exceptions import NoBackChannelError
 from mcp.shared.inbound import (

src/mcp/server/_transport_security_middleware.py

@@ -0,0 +1,100 @@
+"""Starlette middleware that enforces `TransportSecuritySettings` on HTTP transports.
+
+Kept apart from `mcp.server.transport_security` (which defines only the
+web-framework-free settings model, so `import mcp.server` and stdio servers
+never load starlette). `mcp.server.transport_security` still re-exports
+`TransportSecurityMiddleware` for existing imports of it.
+"""
+
+import logging
+
+from starlette.requests import Request
+from starlette.responses import Response
+
+from mcp.server.transport_security import TransportSecuritySettings
+
+logger = logging.getLogger(__name__)
+
+
+# TODO(Marcelo): This should be a proper ASGI middleware. I'm sad to see this.
+class TransportSecurityMiddleware:
+    """Middleware to enforce DNS rebinding protection for MCP transport endpoints."""
+
+    def __init__(self, settings: TransportSecuritySettings | None = None):
+        # If not specified, disable DNS rebinding protection by default for backwards compatibility
+        self.settings = settings or TransportSecuritySettings(enable_dns_rebinding_protection=False)
+
+    def _validate_host(self, host: str | None) -> bool:
+        """Validate the Host header against allowed values."""
+        if not host:
+            logger.warning("Missing Host header in request")
+            return False
+
+        # Check exact match first
+        if host in self.settings.allowed_hosts:
+            return True
+
+        # Check wildcard port patterns
+        for allowed in self.settings.allowed_hosts:
+            if allowed.endswith(":*"):
+                # Extract base host from pattern
+                base_host = allowed[:-2]
+                # Check if the actual host starts with base host and has a port
+                if host.startswith(base_host + ":"):
+                    return True
+
+        logger.warning(f"Invalid Host header: {host}")
+        return False
+
+    def _validate_origin(self, origin: str | None) -> bool:
+        """Validate the Origin header against allowed values."""
+        # Origin can be absent for same-origin requests
+        if not origin:
+            return True
+
+        # Check exact match first
+        if origin in self.settings.allowed_origins:
+            return True
+
+        # Check wildcard port patterns
+        for allowed in self.settings.allowed_origins:
+            if allowed.endswith(":*"):
+                # Extract base origin from pattern
+                base_origin = allowed[:-2]
+                # Check if the actual origin starts with base origin and has a port
+                if origin.startswith(base_origin + ":"):
+                    return True
+
+        logger.warning(f"Invalid Origin header: {origin}")
+        return False
+
+    def _validate_content_type(self, content_type: str | None) -> bool:
+        """Validate the Content-Type header for POST requests."""
+        return content_type is not None and content_type.lower().startswith("application/json")
+
+    async def validate_request(self, request: Request, is_post: bool = False) -> Response | None:
+        """Validate request headers for DNS rebinding protection.
+
+        Returns None if validation passes, or an error Response if validation fails.
+        """
+        # Always validate Content-Type for POST requests
+        if is_post:
+            content_type = request.headers.get("content-type")
+            if not self._validate_content_type(content_type):
+                return Response("Invalid Content-Type header", status_code=400)
+
+        # Skip remaining validation if DNS rebinding protection is disabled
+        if not self.settings.enable_dns_rebinding_protection:
+            return None
+
+        # Validate Host header
+        host = request.headers.get("host")
+        if not self._validate_host(host):
+            return Response("Invalid Host header", status_code=421)
+
+        # Validate Origin header
+        origin = request.headers.get("origin")
+        if not self._validate_origin(origin):
+            return Response("Invalid Origin header", status_code=403)
+
+        return None

src/mcp/server/event_store.py

@@ -0,0 +1,65 @@
+"""Event storage interface for streamable-HTTP resumability.
+
+`EventStore` and its supporting types are the contract a resumable
+streamable-HTTP server plugs a store into (see `mcp.server.streamable_http`,
+which re-exports every name here). They are defined in this transport-agnostic
+module - no web framework imported - so servers and application code can name
+and type against them without loading the HTTP stack.
+"""
+
+from abc import ABC, abstractmethod
+from collections.abc import Awaitable, Callable
+from dataclasses import dataclass
+
+from mcp_types import JSONRPCMessage
+
+__all__ = ["EventCallback", "EventId", "EventMessage", "EventStore", "StreamId"]
+
+# Type aliases
+StreamId = str
+EventId = str
+
+
+@dataclass
+class EventMessage:
+    """A JSONRPCMessage with an optional event ID for stream resumability."""
+
+    message: JSONRPCMessage
+    event_id: str | None = None
+
+
+EventCallback = Callable[[EventMessage], Awaitable[None]]
+
+
+class EventStore(ABC):
+    """Interface for resumability support via event storage."""
+
+    @abstractmethod
+    async def store_event(self, stream_id: StreamId, message: JSONRPCMessage | None) -> EventId:
+        """Stores an event for later retrieval.
+
+        Args:
+            stream_id: ID of the stream the event belongs to
+            message: The JSON-RPC message to store, or None for priming events
+
+        Returns:
+            The generated event ID for the stored event.
+        """
+        pass  # pragma: no cover
+
+    @abstractmethod
+    async def replay_events_after(
+        self,
+        last_event_id: EventId,
+        send_callback: EventCallback,
+    ) -> StreamId | None:
+        """Replays events that occurred after the specified event ID.
+
+        Args:
+            last_event_id: The ID of the last event the client received
+            send_callback: A callback function to send events to the client
+
+        Returns:
+            The stream ID of the replayed events, or None if no events were found.
+        """
+        pass  # pragma: no cover

src/mcp/server/lowlevel/server.py

@@ -50,30 +50,29 @@ async def main():
 from pydantic import BaseModel
 from typing_extensions import TypeVar, deprecated
 
+import mcp
 from mcp.server._http_defaults import DEFAULT_MAX_REQUEST_BODY_SIZE
 from mcp.server._otel import OpenTelemetryMiddleware
 from mcp.server.caching import CacheableMethod, CacheHint, validate_cache_hints
 from mcp.server.context import HandlerResult, ServerMiddleware, ServerRequestContext
+from mcp.server.event_store import EventStore
 from mcp.server.models import InitializationOptions
 from mcp.server.runner import serve_dual_era_loop
+from mcp.server.transport_security import TransportSecuritySettings
 from mcp.shared._stream_protocols import ReadStream, WriteStream
 from mcp.shared.exceptions import MCPDeprecationWarning
 from mcp.shared.message import SessionMessage
 
 if TYPE_CHECKING:
-    # HTTP transport and auth types appear only in `streamable_http_app`'s
-    # signature and in narrowed attributes. The runtime imports live inside
-    # that method, so `import mcp.server` (and every stdio server) never
-    # loads the HTTP stack: starlette, sse_starlette, uvicorn.
+    # Starlette types appear only in `streamable_http_app`'s signature; the
+    # runtime imports live inside that method, so `import mcp.server` (and
+    # every stdio server) never loads the HTTP stack: starlette,
+    # sse_starlette, uvicorn. The auth and session-manager annotations are
+    # spelled through the lazy `mcp.server` namespace instead (below), so
+    # `typing.get_type_hints` still resolves them on demand at runtime.
     from starlette.applications import Starlette
     from starlette.routing import Route
 
-    from mcp.server.auth.provider import OAuthAuthorizationServerProvider, TokenVerifier
-    from mcp.server.auth.settings import AuthSettings
-    from mcp.server.streamable_http import EventStore
-    from mcp.server.streamable_http_manager import StreamableHTTPSessionManager
-    from mcp.server.transport_security import TransportSecuritySettings
-
 logger = logging.getLogger(__name__)
 
 LifespanResultT = TypeVar("LifespanResultT", default=Any)
@@ -425,7 +424,7 @@ def __init__(
         self.lifespan = lifespan
         self._request_handlers: dict[str, HandlerEntry[LifespanResultT]] = {}
         self._notification_handlers: dict[str, HandlerEntry[LifespanResultT]] = {}
-        self._session_manager: StreamableHTTPSessionManager | None = None
+        self._session_manager: mcp.server.streamable_http_manager.StreamableHTTPSessionManager | None = None
         # Context-tier middleware: wraps every inbound request (including
         # `initialize`, lookup, validation, handler) with
         # `(ctx, call_next)`. Applied in `ServerRunner._on_request`.
@@ -674,7 +673,7 @@ async def _handle_discover(
         )
 
     @property
-    def session_manager(self) -> StreamableHTTPSessionManager:
+    def session_manager(self) -> mcp.server.streamable_http_manager.StreamableHTTPSessionManager:
         """Get the StreamableHTTP session manager.
 
         Raises:
@@ -727,9 +726,9 @@ def streamable_http_app(
         max_request_body_size: int = DEFAULT_MAX_REQUEST_BODY_SIZE,
         transport_security: TransportSecuritySettings | None = None,
         host: str = "127.0.0.1",
-        auth: AuthSettings | None = None,
-        token_verifier: TokenVerifier | None = None,
-        auth_server_provider: OAuthAuthorizationServerProvider[Any, Any, Any] | None = None,
+        auth: mcp.server.auth.settings.AuthSettings | None = None,
+        token_verifier: mcp.server.auth.provider.TokenVerifier | None = None,
+        auth_server_provider: mcp.server.auth.provider.OAuthAuthorizationServerProvider[Any, Any, Any] | None = None,
         custom_starlette_routes: list[Route] | None = None,
         debug: bool = False,
     ) -> Starlette:
@@ -751,7 +750,6 @@ def streamable_http_app(
             create_protected_resource_routes,
         )
         from mcp.server.streamable_http_manager import StreamableHTTPASGIApp, StreamableHTTPSessionManager
-        from mcp.server.transport_security import TransportSecuritySettings
 
         # Auto-enable DNS rebinding protection for localhost (IPv4 and IPv6)
         if transport_security is None and host in ("127.0.0.1", "localhost", "::1"):

src/mcp/server/mcpserver/server.py

@@ -47,11 +47,16 @@
 from pydantic import BaseModel
 from pydantic.networks import AnyUrl
 
+import mcp
 from mcp.server._http_defaults import DEFAULT_MAX_REQUEST_BODY_SIZE
-from mcp.server.auth.provider import OAuthAuthorizationServerProvider, ProviderTokenVerifier, TokenVerifier
-from mcp.server.auth.settings import AuthSettings
+
+# `MCPServer.__init__`'s auth-provider annotations are spelled through the lazy
+# `mcp.server.auth` namespace, so this module never imports the OAuth provider
+# stack (it loads with the first auth-enabled server instead).
+from mcp.server.auth.settings import AuthSettings  # light (pydantic-only); the Settings model needs it
 from mcp.server.caching import CacheableMethod, CacheHint
 from mcp.server.context import HandlerResult, ServerMiddleware, ServerRequestContext
+from mcp.server.event_store import EventStore
 from mcp.server.extension import (
     Extension,
     MethodBinding,
@@ -78,25 +83,24 @@
 from mcp.server.request_state import RequestStateBoundary, RequestStateSecurity
 from mcp.server.stdio import stdio_server
 from mcp.server.subscriptions import InMemorySubscriptionBus, ListenHandler, SubscriptionBus
+from mcp.server.transport_security import TransportSecuritySettings
 from mcp.shared.exceptions import MCPError
 from mcp.shared.uri_template import UriTemplate
 
 if TYPE_CHECKING:
-    # HTTP transport types appear only in the SSE / streamable-HTTP methods'
+    # Starlette types appear only in the SSE / streamable-HTTP methods'
     # signatures and in narrowed attributes. Their runtime imports live inside
     # those methods (and `custom_route`), so `import mcp.server.mcpserver`
     # (and every stdio server) never loads starlette, sse_starlette or
-    # uvicorn - only building an HTTP app pays for the web stack, once.
+    # uvicorn - only building an HTTP app pays for the web stack, once. The
+    # session-manager annotation is spelled through the lazy `mcp.server`
+    # namespace instead, so `typing.get_type_hints` resolves it on demand.
     from starlette.applications import Starlette
     from starlette.requests import Request
     from starlette.responses import Response
     from starlette.routing import Route
     from starlette.types import Receive, Scope, Send
 
-    from mcp.server.streamable_http import EventStore
-    from mcp.server.streamable_http_manager import StreamableHTTPSessionManager
-    from mcp.server.transport_security import TransportSecuritySettings
-
 logger = get_logger(__name__)
 
 _CallableT = TypeVar("_CallableT", bound=Callable[..., Any])
@@ -158,8 +162,8 @@ def __init__(
         website_url: str | None = None,
         icons: list[Icon] | None = None,
         version: str = "",
-        auth_server_provider: OAuthAuthorizationServerProvider[Any, Any, Any] | None = None,
-        token_verifier: TokenVerifier | None = None,
+        auth_server_provider: mcp.server.auth.provider.OAuthAuthorizationServerProvider[Any, Any, Any] | None = None,
+        token_verifier: mcp.server.auth.provider.TokenVerifier | None = None,
         *,
         tools: list[Tool] | None = None,
         resources: list[Resource] | None = None,
@@ -250,6 +254,9 @@ def __init__(
 
         # Create token verifier from provider if needed (backwards compatibility)
         if auth_server_provider and not token_verifier:
+            # The OAuth provider stack loads only for a server configured with one.
+            from mcp.server.auth.provider import ProviderTokenVerifier
+
             self._token_verifier = ProviderTokenVerifier(auth_server_provider)
         self._custom_starlette_routes: list[Route] = []
 
@@ -301,7 +308,7 @@ def version(self) -> str:
         return self._lowlevel_server.version
 
     @property
-    def session_manager(self) -> StreamableHTTPSessionManager:
+    def session_manager(self) -> mcp.server.streamable_http_manager.StreamableHTTPSessionManager:
         """Get the StreamableHTTP session manager.
 
         This is exposed to enable advanced use cases like mounting multiple
@@ -1118,7 +1125,6 @@ def sse_app(
         from mcp.server.auth.middleware.auth_context import AuthContextMiddleware
         from mcp.server.auth.middleware.bearer_auth import BearerAuthBackend, RequireAuthMiddleware
         from mcp.server.sse import SseServerTransport
-        from mcp.server.transport_security import TransportSecuritySettings
 
         # Auto-enable DNS rebinding protection for localhost (IPv4 and IPv6)
         if transport_security is None and host in ("127.0.0.1", "localhost", "::1"):

src/mcp/server/sse.py

@@ -50,11 +50,9 @@ async def handle_sse(request):
 from starlette.responses import Response
 from starlette.types import Receive, Scope, Send
 
+from mcp.server._transport_security_middleware import TransportSecurityMiddleware
 from mcp.server.auth.middleware.bearer_auth import AuthenticatedUser, AuthorizationContext, authorization_context
-from mcp.server.transport_security import (
-    TransportSecurityMiddleware,
-    TransportSecuritySettings,
-)
+from mcp.server.transport_security import TransportSecuritySettings
 from mcp.shared._context_streams import ContextSendStream, create_context_streams
 from mcp.shared.message import ServerMessageMetadata, SessionMessage