refactor(auth): decouple core server from auth module for tenant extraction

Move tenant identification to a transport-agnostic contextvar
(tenant_id_var) in the shared layer, removing the hard dependency from
lowlevel/server.py on the auth middleware module.

AuthContextMiddleware now sets tenant_id_var alongside auth_context_var,
and the core server reads from the shared contextvar instead of calling
get_tenant_id() from the auth module. This keeps the dependency direction
correct (auth → shared, server → shared) and allows other transports to
set tenant_id_var through their own mechanisms.

Author: andylim-duo

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

@@ -4,6 +4,7 @@
 
 from mcp.server.auth.middleware.bearer_auth import AuthenticatedUser
 from mcp.server.auth.provider import AccessToken
+from mcp.shared._context import tenant_id_var
 
 # Create a contextvar to store the authenticated user
 # The default is None, indicating no authenticated user is present
@@ -46,11 +47,15 @@ async def __call__(self, scope: Scope, receive: Receive, send: Send):
         user = scope.get("user")
         if isinstance(user, AuthenticatedUser):
             # Set the authenticated user in the contextvar
-            token = auth_context_var.set(user)
+            auth_token = auth_context_var.set(user)
+            # Propagate tenant_id to the transport-agnostic contextvar
+            tenant_id = user.access_token.tenant_id if user.access_token else None
+            tenant_token = tenant_id_var.set(tenant_id)
             try:
                 await self.app(scope, receive, send)
             finally:
-                auth_context_var.reset(token)
+                tenant_id_var.reset(tenant_token)
+                auth_context_var.reset(auth_token)
         else:
             # No authenticated user, just process the request
             await self.app(scope, receive, send)

src/mcp/server/lowlevel/server.py

@@ -52,7 +52,7 @@ async def main():
 from typing_extensions import TypeVar
 
 from mcp import types
-from mcp.server.auth.middleware.auth_context import AuthContextMiddleware, get_tenant_id
+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
@@ -65,6 +65,7 @@ async def main():
 from mcp.server.streamable_http import EventStore
 from mcp.server.streamable_http_manager import StreamableHTTPASGIApp, StreamableHTTPSessionManager
 from mcp.server.transport_security import TransportSecuritySettings
+from mcp.shared._context import tenant_id_var
 from mcp.shared.exceptions import MCPError
 from mcp.shared.message import ServerMessageMetadata, SessionMessage
 from mcp.shared.session import RequestResponder
@@ -451,7 +452,7 @@ async def _handle_request(
                 task_metadata = None
                 if hasattr(req, "params") and req.params is not None:
                     task_metadata = getattr(req.params, "task", None)
-                tenant_id = get_tenant_id()
+                tenant_id = tenant_id_var.get()
                 if tenant_id is not None and session.tenant_id is None:
                     session.tenant_id = tenant_id
                 ctx = ServerRequestContext(
@@ -499,7 +500,7 @@ async def _handle_notification(
             try:
                 client_capabilities = session.client_params.capabilities if session.client_params else None
                 task_support = self._experimental_handlers.task_support if self._experimental_handlers else None
-                tenant_id = get_tenant_id()
+                tenant_id = tenant_id_var.get()
                 if tenant_id is not None and session.tenant_id is None:
                     session.tenant_id = tenant_id
                 ctx = ServerRequestContext(

src/mcp/shared/_context.py

@@ -1,5 +1,6 @@
 """Request context for MCP handlers."""
 
+import contextvars
 from dataclasses import dataclass
 from typing import Any, Generic
 
@@ -8,6 +9,11 @@
 from mcp.shared.session import BaseSession
 from mcp.types import RequestId, RequestParamsMeta
 
+# Transport-agnostic contextvar for tenant identification.
+# Set by the transport layer (e.g., AuthContextMiddleware for HTTP+OAuth).
+# Read by the core server to populate RequestContext.tenant_id.
+tenant_id_var = contextvars.ContextVar[str | None]("tenant_id", default=None)
+
 SessionT = TypeVar("SessionT", bound=BaseSession[Any, Any, Any, Any, Any])
 
 

tests/server/auth/middleware/test_auth_context.py

@@ -13,6 +13,7 @@
 )
 from mcp.server.auth.middleware.bearer_auth import AuthenticatedUser
 from mcp.server.auth.provider import AccessToken
+from mcp.shared._context import tenant_id_var
 
 
 class MockApp:
@@ -169,6 +170,60 @@ async def send(message: Message) -> None:  # pragma: no cover
     assert get_tenant_id() is None
 
 
+@pytest.mark.anyio
+async def test_middleware_sets_tenant_id_var(access_token_with_tenant: AccessToken):
+    """Test AuthContextMiddleware populates the transport-agnostic tenant_id_var."""
+    user = AuthenticatedUser(access_token_with_tenant)
+    scope: Scope = {"type": "http", "user": user}
+
+    observed_tenant_id: str | None = None
+
+    class CheckApp:
+        async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
+            nonlocal observed_tenant_id
+            observed_tenant_id = tenant_id_var.get()
+
+    middleware = AuthContextMiddleware(CheckApp())
+
+    async def receive() -> Message:  # pragma: no cover
+        return {"type": "http.request"}
+
+    async def send(message: Message) -> None:  # pragma: no cover
+        pass
+
+    await middleware(scope, receive, send)
+
+    assert observed_tenant_id == "tenant-abc"
+    # Verify contextvar is reset after middleware
+    assert tenant_id_var.get() is None
+
+
+@pytest.mark.anyio
+async def test_middleware_sets_tenant_id_var_none_without_tenant(valid_access_token: AccessToken):
+    """Test AuthContextMiddleware sets tenant_id_var to None when token has no tenant."""
+    user = AuthenticatedUser(valid_access_token)
+    scope: Scope = {"type": "http", "user": user}
+
+    observed_tenant_id: str | None = "sentinel"
+
+    class CheckApp:
+        async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
+            nonlocal observed_tenant_id
+            observed_tenant_id = tenant_id_var.get()
+
+    middleware = AuthContextMiddleware(CheckApp())
+
+    async def receive() -> Message:  # pragma: no cover
+        return {"type": "http.request"}
+
+    async def send(message: Message) -> None:  # pragma: no cover
+        pass
+
+    await middleware(scope, receive, send)
+
+    assert observed_tenant_id is None
+
+
 @pytest.mark.anyio
 async def test_get_tenant_id_without_tenant(valid_access_token: AccessToken):
     """Test get_tenant_id returns None when auth context has no tenant."""

tests/server/test_multi_tenancy_session.py

@@ -14,14 +14,18 @@
 from mcp.server.experimental.request_context import Experimental
 from mcp.server.models import InitializationOptions
 from mcp.server.session import ServerSession
-from mcp.shared._context import RequestContext
+from mcp.shared._context import RequestContext, tenant_id_var
 from mcp.shared.message import SessionMessage
 from mcp.shared.session import BaseSession
 from mcp.types import ListToolsResult, NotificationParams, PaginatedRequestParams, ServerCapabilities
 
 
 def _simulate_tenant_binding(session: ServerSession, tenant_id_value: str) -> None:
-    """Simulate the set-once tenant binding logic from lowlevel/server.py."""
+    """Simulate the set-once tenant binding logic from lowlevel/server.py.
+
+    Sets both auth_context_var (as AuthContextMiddleware does) and tenant_id_var
+    (the transport-agnostic contextvar that the server reads).
+    """
     access_token = AccessToken(
         token=f"token-{tenant_id_value}",
         client_id="client",
@@ -30,13 +34,15 @@ def _simulate_tenant_binding(session: ServerSession, tenant_id_value: str) -> No
         tenant_id=tenant_id_value,
     )
     user = AuthenticatedUser(access_token)
-    context_token = auth_context_var.set(user)
+    auth_token = auth_context_var.set(user)
+    tenant_token = tenant_id_var.set(tenant_id_value)
     try:
-        tenant_id = get_tenant_id()
+        tenant_id = tenant_id_var.get()
         if tenant_id is not None and session.tenant_id is None:
             session.tenant_id = tenant_id
     finally:
-        auth_context_var.reset(context_token)
+        tenant_id_var.reset(tenant_token)
+        auth_context_var.reset(auth_token)
 
 
 @pytest.fixture
@@ -269,18 +275,20 @@ async def simulate_request(tenant_id: str, request_key: str) -> None:
         )
         user = AuthenticatedUser(access_token)
 
-        # Set the auth context - this is what AuthContextMiddleware does
-        context_token = auth_context_var.set(user)
+        # Set both contextvars - this is what AuthContextMiddleware does
+        auth_token = auth_context_var.set(user)
+        tenant_token = tenant_id_var.set(tenant_id)
         try:
             # Yield control to allow other tasks to run. This is the critical
             # point where context leakage could occur if isolation is broken.
             await anyio.sleep(0.01)
 
             # Read back the tenant_id - should still be our tenant, not the other
-            results[request_key] = get_tenant_id()
+            results[request_key] = tenant_id_var.get()
         finally:
             # Always reset the context (mirrors middleware behavior)
-            auth_context_var.reset(context_token)
+            tenant_id_var.reset(tenant_token)
+            auth_context_var.reset(auth_token)
 
     # Run both requests concurrently using a task group
     async with anyio.create_task_group() as tg:
@@ -362,12 +370,14 @@ async def handle_list_tools(ctx: ServerRequestContext, params: PaginatedRequestP
         tenant_id="tenant-e2e",
     )
     user = AuthenticatedUser(access_token)
-    token = auth_context_var.set(user)
+    auth_token = auth_context_var.set(user)
+    tenant_token = tenant_id_var.set("tenant-e2e")
     try:
         async with Client(server) as client:
             await client.list_tools()
     finally:
-        auth_context_var.reset(token)
+        tenant_id_var.reset(tenant_token)
+        auth_context_var.reset(auth_token)
 
     assert captured_ctx_tenant == "tenant-e2e"
     assert captured_session_tenant == "tenant-e2e"
@@ -399,13 +409,15 @@ async def handle_roots_list_changed(ctx: ServerRequestContext, params: Notificat
         tenant_id="tenant-notify",
     )
     user = AuthenticatedUser(access_token)
-    token = auth_context_var.set(user)
+    auth_token = auth_context_var.set(user)
+    tenant_token = tenant_id_var.set("tenant-notify")
     try:
         async with Client(server) as client:
             await client.session.send_roots_list_changed()
             with anyio.fail_after(5):
                 await notification_received.wait()
     finally:
-        auth_context_var.reset(token)
+        tenant_id_var.reset(tenant_token)
+        auth_context_var.reset(auth_token)
 
     assert notification_tenant == "tenant-notify"