Make server-side imports pay-for-what-you-use

The HTTP transport stack (starlette, sse_starlette, uvicorn) is no
longer imported by the transport-agnostic server modules at module
import: the lowlevel Server and MCPServer import the web stack inside
streamable_http_app() / sse_app() / custom_route(), with
annotation-only names moved under TYPE_CHECKING, so a stdio server
never loads it. The request access-token contextvar and
get_access_token, which the request-state boundary and handlers read
regardless of transport, move to a starlette-free module,
mcp.server.auth.access_token, and are re-exported from
mcp.server.auth.middleware.auth_context so the existing import path
keeps working; the boundary now reads the principal without the HTTP
stack. opentelemetry-api is imported on the first span instead of at
import (cached in module globals; otel_span takes the span kind by
name), and httpx2 is imported only when an HttpResource is read. The
streamable-HTTP request-body-size default moves to a leaf module so
the servers' signatures do not import the transport.

src/mcp/server/_http_defaults.py

@@ -0,0 +1,13 @@
+"""Streamable HTTP defaults shared by the transport and the servers' signatures.
+
+A leaf module with no third-party imports: `mcp.server.lowlevel` and
+`mcp.server.mcpserver` use these as parameter defaults, and importing them from
+the HTTP transport modules (which need starlette) would drag the HTTP stack
+into every stdio server at import time. `mcp.server.streamable_http_manager`
+re-exports the constant under its documented import path.
+"""
+
+from typing import Final
+
+DEFAULT_MAX_REQUEST_BODY_SIZE: Final = 4 * 1024 * 1024
+"""Default maximum Streamable HTTP request body size in bytes (4 MiB)."""

src/mcp/server/_otel.py

@@ -3,11 +3,10 @@ from __future__ import annotations
 from typing import Any
 
 from mcp_types import INVALID_PARAMS, CallToolResult
-from opentelemetry.trace import SpanKind, StatusCode
 from pydantic import ValidationError
 
 from mcp.server.context import CallNext, HandlerResult, ServerMiddleware, ServerRequestContext
-from mcp.shared._otel import extract_trace_context, otel_span
+from mcp.shared._otel import extract_trace_context, otel_span, set_span_error
 from mcp.shared.exceptions import MCPError
 
 
@@ -34,7 +33,7 @@ class OpenTelemetryMiddleware(ServerMiddleware[Any]):
 
         with otel_span(
             name=f"{ctx.method}{f' {target}' if target else ''}",
-            kind=SpanKind.SERVER,
+            kind="server",
             attributes=attributes,
             context=extract_trace_context(ctx.meta),
             record_exception=False,
@@ -45,18 +44,18 @@ class OpenTelemetryMiddleware(ServerMiddleware[Any]):
             except MCPError as e:
                 code = str(e.error.code)
                 span.set_attributes({"error.type": code, "rpc.response.status_code": code})
-                span.set_status(StatusCode.ERROR, e.error.message)
+                set_span_error(span, e.error.message)
                 raise
             except ValidationError:
                 # Mirror the sanitized wire response; pydantic messages carry client input.
                 code = str(INVALID_PARAMS)
                 span.set_attributes({"error.type": code, "rpc.response.status_code": code})
-                span.set_status(StatusCode.ERROR, "Invalid request parameters")
+                set_span_error(span, "Invalid request parameters")
                 raise
             except Exception as e:
                 span.set_attribute("error.type", type(e).__qualname__)
                 span.record_exception(e)
-                span.set_status(StatusCode.ERROR, str(e))
+                set_span_error(span, str(e))
                 raise
             if ctx.method == "tools/call":
                 # Tool errors are detected pre-serialization, so only shapes that reach the wire as an error
@@ -66,7 +65,7 @@ class OpenTelemetryMiddleware(ServerMiddleware[Any]):
                 match result:
                     case CallToolResult(is_error=True) | {"isError": True}:
                         span.set_attribute("error.type", "tool_error")
-                        span.set_status(StatusCode.ERROR)
+                        set_span_error(span)
                     case _:
                         pass
             return result

src/mcp/server/auth/access_token.py

@@ -0,0 +1,32 @@
+"""The access token of the request being served, exposed via a contextvar.
+
+This module is deliberately transport-agnostic: it imports no HTTP framework,
+so `mcp.server.request_state` (and any tool handler) can read the caller's
+token without loading the web stack. On HTTP transports the contextvar is
+populated by `mcp.server.auth.middleware.auth_context.AuthContextMiddleware`,
+which also re-exports both names under their long-standing import path.
+"""
+
+import contextvars
+from typing import TYPE_CHECKING
+
+from mcp.server.auth.provider import AccessToken
+
+if TYPE_CHECKING:
+    from mcp.server.auth.middleware.bearer_auth import AuthenticatedUser
+
+# Create a contextvar to store the authenticated user
+# The default is None, indicating no authenticated user is present
+auth_context_var: contextvars.ContextVar["AuthenticatedUser | None"] = contextvars.ContextVar(
+    "auth_context", default=None
+)
+
+
+def get_access_token() -> AccessToken | None:
+    """Get the access token from the current context.
+
+    Returns:
+        The access token if an authenticated user is available, None otherwise.
+    """
+    auth_user = auth_context_var.get()
+    return auth_user.access_token if auth_user else None

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

@@ -1,23 +1,11 @@
-import contextvars
-
 from starlette.types import ASGIApp, Receive, Scope, Send
 
+# The contextvar and its accessor are defined in a transport-agnostic module
+# (no starlette) so request-state code can read the token without loading the
+# web stack; they are re-exported here under their long-standing import path.
+from mcp.server.auth.access_token import auth_context_var as auth_context_var
+from mcp.server.auth.access_token import get_access_token as get_access_token
 from mcp.server.auth.middleware.bearer_auth import AuthenticatedUser
-from mcp.server.auth.provider import AccessToken
-
-# Create a contextvar to store the authenticated user
-# The default is None, indicating no authenticated user is present
-auth_context_var = contextvars.ContextVar[AuthenticatedUser | None]("auth_context", default=None)
-
-
-def get_access_token() -> AccessToken | None:
-    """Get the access token from the current context.
-
-    Returns:
-        The access token if an authenticated user is available, None otherwise.
-    """
-    auth_user = auth_context_var.get()
-    return auth_user.access_token if auth_user else None
 
 
 class AuthContextMiddleware:

src/mcp/server/lowlevel/server.py

@@ -43,38 +43,37 @@ from collections.abc import AsyncIterator, Awaitable, Callable, Mapping
 from contextlib import AbstractAsyncContextManager, asynccontextmanager
 from dataclasses import dataclass
 from functools import cached_property
-from typing import Any, Generic, overload
+from typing import TYPE_CHECKING, Any, Generic, overload
 
 import mcp_types as types
 from mcp_types.version import MODERN_PROTOCOL_VERSIONS
 from pydantic import BaseModel
-from starlette.applications import Starlette
-from starlette.middleware import Middleware
-from starlette.middleware.authentication import AuthenticationMiddleware
-from starlette.routing import Mount, Route
 from typing_extensions import TypeVar, deprecated
 
+from mcp.server._http_defaults import DEFAULT_MAX_REQUEST_BODY_SIZE
 from mcp.server._otel import OpenTelemetryMiddleware
-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 AuthSettings
 from mcp.server.caching import CacheableMethod, CacheHint, validate_cache_hints
 from mcp.server.context import HandlerResult, ServerMiddleware, ServerRequestContext
 from mcp.server.models import InitializationOptions
 from mcp.server.runner import serve_dual_era_loop
-from mcp.server.streamable_http import EventStore
-from mcp.server.streamable_http_manager import (
-    DEFAULT_MAX_REQUEST_BODY_SIZE,
-    StreamableHTTPASGIApp,
-    StreamableHTTPSessionManager,
-)
-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.
+    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)
@@ -735,6 +734,25 @@ class Server(Generic[LifespanResultT]):
         debug: bool = False,
     ) -> Starlette:
         """Return an instance of the StreamableHTTP server app."""
+        # The HTTP transport stack (starlette, plus this SDK's HTTP transport
+        # and auth ASGI modules) is imported here rather than at module top so
+        # that `import mcp.server` and stdio servers never load starlette,
+        # sse_starlette or uvicorn: only building an HTTP app pays for it, once.
+        from starlette.applications import Starlette
+        from starlette.middleware import Middleware
+        from starlette.middleware.authentication import AuthenticationMiddleware
+        from starlette.routing import Mount, Route
+
+        from mcp.server.auth.middleware.auth_context import AuthContextMiddleware
+        from mcp.server.auth.middleware.bearer_auth import BearerAuthBackend, RequireAuthMiddleware
+        from mcp.server.auth.routes import (
+            build_resource_metadata_url,
+            create_auth_routes,
+            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"):
             transport_security = TransportSecuritySettings(

src/mcp/server/mcpserver/resources/types.py

@@ -10,7 +10,6 @@ from typing import Any
 
 import anyio
 import anyio.to_thread
-import httpx2
 import pydantic
 import pydantic_core
 from mcp_types import Annotations, Icon, InputRequiredResult
@@ -197,9 +196,14 @@ class HttpResource(Resource):
     url: str = Field(description="URL to fetch content from")
     mime_type: str = Field(default="application/json", description="MIME type of the resource content")
 
-    async def read(self) -> str | bytes:
+    async def read(self) -> str | bytes:  # pragma: no cover
         """Read the HTTP content."""
-        async with httpx2.AsyncClient() as client:  # pragma: no cover
+        # httpx2 is imported here rather than at module top: this is the only
+        # resource type that needs the HTTP client stack, and a server that
+        # never registers an HttpResource should not pay for it at import time.
+        import httpx2
+
+        async with httpx2.AsyncClient() as client:
             response = await client.get(self.url)
             response.raise_for_status()
             return response.text

src/mcp/server/mcpserver/server.py

@@ -6,7 +6,7 @@ import base64
 import inspect
 from collections.abc import AsyncIterator, Awaitable, Callable, Iterable, Mapping, Sequence
 from contextlib import AbstractAsyncContextManager, asynccontextmanager
-from typing import Any, Generic, Literal, TypeVar, overload
+from typing import TYPE_CHECKING, Any, Generic, Literal, TypeVar, overload
 
 import anyio
 import pydantic_core
@@ -46,16 +46,8 @@ from mcp_types import ResourceTemplate as MCPResourceTemplate
 from mcp_types import Tool as MCPTool
 from pydantic import BaseModel
 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
 
-from mcp.server.auth.middleware.auth_context import AuthContextMiddleware
-from mcp.server.auth.middleware.bearer_auth import BearerAuthBackend, RequireAuthMiddleware
+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
 from mcp.server.caching import CacheableMethod, CacheHint
@@ -84,15 +76,27 @@ from mcp.server.mcpserver.tools import Tool, ToolManager
 from mcp.server.mcpserver.utilities.context_injection import find_context_parameter
 from mcp.server.mcpserver.utilities.logging import configure_logging, get_logger
 from mcp.server.request_state import RequestStateBoundary, RequestStateSecurity
-from mcp.server.sse import SseServerTransport
 from mcp.server.stdio import stdio_server
-from mcp.server.streamable_http import EventStore
-from mcp.server.streamable_http_manager import DEFAULT_MAX_REQUEST_BODY_SIZE, StreamableHTTPSessionManager
 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'
+    # 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.
+    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])
@@ -1005,6 +1009,10 @@ class MCPServer(Generic[LifespanResultT]):
             ```
         """
 
+        # A custom route is an HTTP feature: starlette is imported here rather
+        # than at module top so stdio servers never load the HTTP stack.
+        from starlette.routing import Route
+
         def decorator(
             func: Callable[[Request], Awaitable[Response]],
         ) -> Callable[[Request], Awaitable[Response]]:
@@ -1097,6 +1105,21 @@ class MCPServer(Generic[LifespanResultT]):
         host: str = "127.0.0.1",
     ) -> Starlette:
         """Return an instance of the SSE server app."""
+        # The SSE transport stack (starlette, sse_starlette, plus this SDK's SSE
+        # transport and auth ASGI modules) is imported here rather than at module
+        # top so that stdio servers never load it: only building an SSE app
+        # pays for it, once.
+        from starlette.applications import Starlette
+        from starlette.middleware import Middleware
+        from starlette.middleware.authentication import AuthenticationMiddleware
+        from starlette.responses import Response
+        from starlette.routing import Mount, Route
+
+        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"):
             transport_security = TransportSecuritySettings(

src/mcp/server/request_state.py

@@ -26,7 +26,7 @@ from cryptography.hazmat.primitives.kdf.hkdf import HKDF
 from mcp_types import INTERNAL_ERROR, INVALID_PARAMS
 from mcp_types.methods import INPUT_REQUIRED_METHODS, is_input_required
 
-from mcp.server.auth.middleware.auth_context import get_access_token
+from mcp.server.auth.access_token import get_access_token
 from mcp.server.auth.provider import principal_components
 from mcp.server.context import CallNext, HandlerResult, ServerRequestContext
 from mcp.shared.exceptions import MCPError

src/mcp/server/streamable_http_manager.py

@@ -6,7 +6,7 @@ import contextlib
 import logging
 from collections import deque
 from collections.abc import AsyncIterator
-from typing import TYPE_CHECKING, Any, Final
+from typing import TYPE_CHECKING, Any
 from uuid import uuid4
 
 import anyio
@@ -18,6 +18,7 @@ from starlette.requests import Request
 from starlette.responses import Response
 from starlette.types import ASGIApp, Message, Receive, Scope, Send
 
+from mcp.server._http_defaults import DEFAULT_MAX_REQUEST_BODY_SIZE as DEFAULT_MAX_REQUEST_BODY_SIZE
 from mcp.server._streamable_http_modern import handle_modern_request
 from mcp.server.auth.middleware.bearer_auth import AuthenticatedUser, AuthorizationContext, authorization_context
 from mcp.server.connection import Connection
@@ -34,9 +35,6 @@ if TYPE_CHECKING:
 
 logger = logging.getLogger(__name__)
 
-DEFAULT_MAX_REQUEST_BODY_SIZE: Final = 4 * 1024 * 1024
-"""Default maximum Streamable HTTP request body size in bytes (4 MiB)."""
-
 
 class StreamableHTTPSessionManager:
     """Manages StreamableHTTP sessions with optional resumability via event store.

src/mcp/shared/_otel.py

@@ -1,33 +1,93 @@
-"""OpenTelemetry helpers for MCP."""
+"""OpenTelemetry helpers for MCP.
+
+`opentelemetry-api` is a hard dependency, but it is imported here on the first
+span rather than at module import: every SDK entry point (client and server,
+over every transport) routes messages through this module, and a tracer that is
+a no-op until an exporter is installed should not add its import cost to
+`import mcp` or to a process that never emits a span. The API modules and the
+tracer are resolved once and cached in module globals; nothing is re-imported
+per message.
+"""
 
 from __future__ import annotations
 
 from collections.abc import Generator, Mapping
 from contextlib import contextmanager
-from typing import Any
+from typing import TYPE_CHECKING, Any, Literal
 
-from opentelemetry.context import Context
-from opentelemetry.propagate import extract, inject
-from opentelemetry.trace import SpanKind, get_current_span, get_tracer
-from opentelemetry.trace.span import Span
+if TYPE_CHECKING:
+    from opentelemetry.context import Context
+    from opentelemetry.trace import SpanKind, Tracer
+    from opentelemetry.trace.span import Span
 
-_tracer = get_tracer("mcp-python-sdk")
+SpanKindName = Literal["client", "server"]
+"""Name of an `opentelemetry.trace.SpanKind` member, resolved once the API loads.
+
+Callers pass the name (or a `SpanKind` member directly) so they need not import
+opentelemetry themselves.
+"""
+
+# The tracer used for every MCP span: `None` until the first span, then the
+# tracer OpenTelemetry late-binds to the global provider (so an exporter
+# configured after import still lights spans up). Assigning it directly is
+# the seam tests use to scope spans to a capturing provider.
+_tracer: Tracer | None = None
+
+# The `opentelemetry.trace` / `opentelemetry.propagate` modules, imported
+# together on first use by `_load` (typed `Any`: they are modules).
+_trace: Any = None
+_propagate: Any = None
+
+
+def _load() -> Any:
+    """Import the OpenTelemetry API, cache it, and return `opentelemetry.trace`.
+
+    Deferred to first use purely for import time (see the module docstring);
+    every later call is served from the module-level cache by `_api`.
+    """
+    global _trace, _propagate
+    from opentelemetry import propagate, trace
+
+    _propagate = propagate
+    _trace = trace
+    return trace
+
+
+def _api() -> Any:
+    """The `opentelemetry.trace` module, importing the OTel API on first use."""
+    return _trace if _trace is not None else _load()
+
+
+def _get_tracer() -> Tracer:
+    """The MCP tracer, created from the global provider on the first span."""
+    global _tracer
+    tracer = _tracer
+    if tracer is None:
+        tracer = _tracer = _api().get_tracer("mcp-python-sdk")
+    return tracer
 
 
 @contextmanager
 def otel_span(
     name: str,
     *,
-    kind: SpanKind,
+    kind: SpanKindName | SpanKind,
     attributes: dict[str, Any] | None = None,
     context: Context | None = None,
     record_exception: bool = True,
     set_status_on_exception: bool = True,
 ) -> Generator[Span]:
     """Create an OTel span."""
-    with _tracer.start_as_current_span(
+    tracer = _get_tracer()
+    if kind == "client":
+        span_kind: SpanKind = _api().SpanKind.CLIENT
+    elif kind == "server":
+        span_kind = _api().SpanKind.SERVER
+    else:
+        span_kind = kind
+    with tracer.start_as_current_span(
         name,
-        kind=kind,
+        kind=span_kind,
         attributes=attributes,
         context=context,
         record_exception=record_exception,
@@ -36,9 +96,15 @@ def otel_span(
         yield span
 
 
+def set_span_error(span: Span, description: str | None = None) -> None:
+    """Mark `span` as errored (`StatusCode.ERROR`), with an optional description."""
+    span.set_status(_api().StatusCode.ERROR, description)
+
+
 def inject_trace_context(meta: dict[str, Any]) -> None:
     """Inject W3C trace context (traceparent/tracestate) into a `_meta` dict."""
-    inject(meta)
+    _api()
+    _propagate.inject(meta)
 
 
 def extract_trace_context(meta: Mapping[str, Any] | None) -> Context | None:
@@ -51,10 +117,11 @@ def extract_trace_context(meta: Mapping[str, Any] | None) -> Context | None:
     """
     if not meta:
         return None
+    trace = _api()
     try:
-        ctx = extract(meta)
+        ctx = _propagate.extract(meta)
     except (ValueError, TypeError):
         return None
-    if not get_current_span(ctx).get_span_context().is_valid:
+    if not trace.get_current_span(ctx).get_span_context().is_valid:
         return None
     return ctx

src/mcp/shared/jsonrpc_dispatcher.py

@@ -32,7 +32,6 @@ from mcp_types import (
     ProgressToken,
     RequestId,
 )
-from opentelemetry.trace import SpanKind
 from pydantic import ValidationError
 from typing_extensions import TypeVar
 
@@ -383,7 +382,7 @@ class JSONRPCDispatcher(Dispatcher[TransportT]):
         try:
             with otel_span(
                 span_name,
-                kind=SpanKind.CLIENT,
+                kind="client",
                 attributes={"mcp.method.name": method, "jsonrpc.request.id": str(request_id)},
             ):
                 # SEP-414: inject W3C trace context; `_meta` stays on the wire even with a no-op tracer.