Load the OAuth provider stack and cryptography with their first user, not the server

Two dependencies still rode along with every `import mcp.server*` for
features most servers never use: the OAuth provider models (via
mcp.shared.auth and urllib.parse, ~10 ms) were imported by the
access-token leaf only for a return annotation and by request_state for
principal decomposition, and cryptography (~10 ms, 20-odd modules) was
imported at request_state's module top for the built-in codec.

The access-token module now spells its return type through the lazy
mcp.server.auth namespace and imports no OAuth model; request_state
resolves the provider's principal_components and the AEAD/KDF primitives
through cached loaders on first use (a codec is built by MCPServer(), so
that is where cryptography now loads - once, at construction, never per
request). The import-cost ratchet adds cryptography to the banned set of
every transport-agnostic server entry point and pins where its deferred
load lands.

Author: maxisbey

src/mcp/server/auth/access_token.py

@@ -1,28 +1,34 @@
 """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.
+This module is deliberately transport- and auth-stack-agnostic: it imports no
+HTTP framework and no OAuth model, so `mcp.server.request_state` (and any tool
+handler) can read the caller's token without loading either. 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.
 """
 
+from __future__ import annotations
+
 import contextvars
 from typing import TYPE_CHECKING
 
-from mcp.server.auth.provider import AccessToken
+import mcp
 
 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_var: contextvars.ContextVar[AuthenticatedUser | None] = contextvars.ContextVar(
     "auth_context", default=None
 )
 
 
-def get_access_token() -> AccessToken | None:
+# The return type is spelled through the lazy `mcp.server` namespace so this
+# module never imports the OAuth provider models, while
+# `typing.get_type_hints(get_access_token)` still resolves it on demand.
+def get_access_token() -> mcp.server.auth.provider.AccessToken | None:
     """Get the access token from the current context.
 
     Returns:

src/mcp/server/request_state.py

@@ -17,20 +17,21 @@
 import time
 from collections.abc import Callable, Mapping, Sequence
 from dataclasses import replace
-from typing import Any, NoReturn, Protocol, cast
+from functools import cache
+from typing import TYPE_CHECKING, Any, NamedTuple, NoReturn, Protocol, cast
 
-from cryptography.exceptions import InvalidTag
-from cryptography.hazmat.primitives.ciphers.aead import AESGCM
-from cryptography.hazmat.primitives.hashes import SHA256
-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.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
 
+if TYPE_CHECKING:
+    from cryptography.hazmat.primitives.ciphers.aead import AESGCM
+
+    from mcp.server.auth.provider import AccessToken
+
 __all__ = [
     "AESGCMRequestStateCodec",
     "InvalidRequestState",
@@ -88,7 +89,7 @@ def authenticated_principal(ctx: ServerRequestContext[Any, Any]) -> str | None:
     token = get_access_token()
     if token is None:
         return None
-    return compact_json(principal_components(token))
+    return compact_json(_principal_components()(token))
 
 
 class RequestStateSecurity:
@@ -179,9 +180,47 @@ def _b64u_decode(text: str) -> bytes:
     return raw
 
 
+class _Crypto(NamedTuple):
+    """The `cryptography` primitives the built-in codec uses, imported together."""
+
+    aesgcm: type[AESGCM]
+    hkdf: Any
+    sha256: Any
+    invalid_tag: type[BaseException]
+
+
+@cache
+def _crypto() -> _Crypto:
+    """Import (once) the AEAD/KDF primitives from `cryptography`.
+
+    Kept off `import mcp.server*`: only a server that seals `requestState` with the
+    built-in codec needs them, so the ~20 modules load with the first codec
+    construction instead of with every server import.
+    """
+    from cryptography.exceptions import InvalidTag
+    from cryptography.hazmat.primitives.ciphers.aead import AESGCM
+    from cryptography.hazmat.primitives.hashes import SHA256
+    from cryptography.hazmat.primitives.kdf.hkdf import HKDF
+
+    return _Crypto(AESGCM, HKDF, SHA256, InvalidTag)
+
+
+@cache
+def _principal_components() -> Callable[[AccessToken], tuple[str, str | None, str | None]]:
+    """Import (once) the auth stack's principal decomposition.
+
+    Kept off `import mcp.server*`: only an authenticated transport reaches it,
+    on the first request that binds request state to a principal.
+    """
+    from mcp.server.auth.provider import principal_components
+
+    return principal_components
+
+
 def _derive_key(secret: bytes) -> bytes:
     """Stretch an operator secret (>= 32 bytes, any format) into the AES-256 key."""
-    return HKDF(algorithm=SHA256(), length=32, salt=None, info=_KDF_INFO).derive(secret)
+    crypto = _crypto()
+    return crypto.hkdf(algorithm=crypto.sha256(), length=32, salt=None, info=_KDF_INFO).derive(secret)
 
 
 class AESGCMRequestStateCodec:
@@ -212,14 +251,15 @@ def __init__(self, keys: Sequence[bytes | bytearray | str]) -> None:
                     f"keys[{i}] is {len(k)} bytes. "
                     'Generate one with: python -c "import secrets; print(secrets.token_hex(32))"'
                 )
+        crypto = self._crypto = _crypto()
         self._ring: dict[bytes, AESGCM] = {}
         self._mint_kid = b""
         for i, secret in enumerate(material):
             key = _derive_key(secret)
             kid = hashlib.sha256(_KID_INFO + key).digest()[:_KID_LEN]
             if kid in self._ring:
                 raise ValueError(f"keys[{i}] duplicates an earlier ring key")
-            self._ring[kid] = AESGCM(key)
+            self._ring[kid] = crypto.aesgcm(key)
             if i == 0:
                 self._mint_kid = kid
 
@@ -244,7 +284,7 @@ def unseal(self, token: str) -> bytes:
             raise InvalidRequestState("unknown key")
         try:
             return aead.decrypt(nonce, sealed, _TOKEN_PREFIX.encode() + kid)
-        except InvalidTag:
+        except self._crypto.invalid_tag:
             raise InvalidRequestState("seal") from None
 
 

tests/test_import_guards.py

@@ -45,8 +45,10 @@
 # stack, and the auth/telemetry dependencies that only server-side code paths use.
 CLIENT_BANNED = ("mcp.server", *HTTP_SERVER_STACK, "opentelemetry", "cryptography", "jwt")
 
-# Everything a transport-agnostic server entry point must never load.
-SERVER_BANNED = (*HTTP_SERVER_STACK, "opentelemetry", "httpx2", "jwt")
+# Everything a transport-agnostic server entry point must never load. `cryptography` backs
+# the built-in request-state codec and loads with the first codec construction (`MCPServer()`
+# builds one), never at import.
+SERVER_BANNED = (*HTTP_SERVER_STACK, "opentelemetry", "httpx2", "jwt", "cryptography")
 
 # Per-version wire-schema packages: loaded only when that protocol version is used.
 WIRE_PACKAGES = ("mcp_types._v2025_11_25", "mcp_types._v2026_07_28")
@@ -146,6 +148,23 @@ def test_first_url_client_is_what_loads_the_http_client_transport():
     assert _hits(after, CLIENT_BANNED) == []
 
 
+def test_request_state_codec_is_what_loads_cryptography():
+    """The deferred load happens where it is used: importing the server does not load
+    `cryptography`; constructing an `MCPServer` (which builds the default request-state
+    codec) is what does."""
+    probe = (
+        "import json, sys\n"
+        "from mcp.server.mcpserver import MCPServer\n"
+        "before = sorted(sys.modules)\n"
+        "MCPServer('probe')\n"
+        "print(json.dumps([before, sorted(sys.modules)]))\n"
+    )
+    before, after = (set(names) for names in json.loads(_run(probe)))
+    assert _hits(before, SERVER_BANNED) == []
+    assert _hits(after, ("cryptography",)) == ["cryptography"]
+    assert _hits(after, HTTP_SERVER_STACK) == []
+
+
 @pytest.mark.parametrize("module_name", _documented_modules())
 def test_documented_module_imports_first_in_a_fresh_interpreter(module_name: str):
     """Every documented module imports cleanly and warning-free as a process's first