From 293a6eeb318eb76e6f4b9d9118b89f0de75a9ea2 Mon Sep 17 00:00:00 2001
From: Andy Lim <aclim@cisco.com>
Date: Wed, 11 Mar 2026 11:24:14 -0400
Subject: [PATCH 01/14] feat(auth): add tenant_id to session and request
 context

Thread tenant_id from authentication tokens through the request
lifecycle to enable tenant-scoped operations in handlers.

Changes:
- Add tenant_id field to base RequestContext (inherited by ServerRequestContext)
- Add get_tenant_id() helper in auth_context module to extract tenant from auth
- Populate tenant_id in both ServerRequestContext instantiation sites in lowlevel/server.py
- Add tenant_id property with getter/setter to ServerSession

This is iteration 2 of the multi-tenancy implementation, building on
the tenant_id field added to auth tokens in iteration 1.
---
 .../server/auth/middleware/auth_context.py    |  10 ++
 src/mcp/server/lowlevel/server.py             |   4 +-
 src/mcp/server/session.py                     |  11 ++
 src/mcp/shared/_context.py                    |   1 +
 .../auth/middleware/test_auth_context.py      |  76 ++++++++
 tests/server/test_multi_tenancy_session.py    | 163 ++++++++++++++++++
 6 files changed, 264 insertions(+), 1 deletion(-)
 create mode 100644 tests/server/test_multi_tenancy_session.py

diff --git a/src/mcp/server/auth/middleware/auth_context.py b/src/mcp/server/auth/middleware/auth_context.py
index 1d34a5546b..c619d81a56 100644
--- a/src/mcp/server/auth/middleware/auth_context.py
+++ b/src/mcp/server/auth/middleware/auth_context.py
@@ -20,6 +20,16 @@ def get_access_token() -> AccessToken | None:
     return auth_user.access_token if auth_user else None
 
 
+def get_tenant_id() -> str | None:
+    """Get the tenant_id from the current authentication context.
+
+    Returns:
+        The tenant_id if an authenticated user with a tenant is available, None otherwise.
+    """
+    access_token = get_access_token()
+    return access_token.tenant_id if access_token else None
+
+
 class AuthContextMiddleware:
     """Middleware that extracts the authenticated user from the request
     and sets it in a contextvar for easy access throughout the request lifecycle.
diff --git a/src/mcp/server/lowlevel/server.py b/src/mcp/server/lowlevel/server.py
index 1c84c86107..8a7a278853 100644
--- a/src/mcp/server/lowlevel/server.py
+++ b/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
+from mcp.server.auth.middleware.auth_context import AuthContextMiddleware, get_tenant_id
 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
@@ -461,6 +461,7 @@ async def _handle_request(
                     meta=message.request_meta,
                     session=session,
                     lifespan_context=lifespan_context,
+                    tenant_id=get_tenant_id(),
                     experimental=Experimental(
                         task_metadata=task_metadata,
                         _client_capabilities=client_capabilities,
@@ -503,6 +504,7 @@ async def _handle_notification(
                 ctx = ServerRequestContext(
                     session=session,
                     lifespan_context=lifespan_context,
+                    tenant_id=get_tenant_id(),
                     experimental=Experimental(
                         task_metadata=None,
                         _client_capabilities=client_capabilities,
diff --git a/src/mcp/server/session.py b/src/mcp/server/session.py
index 759d2131a1..fde9836b84 100644
--- a/src/mcp/server/session.py
+++ b/src/mcp/server/session.py
@@ -76,6 +76,7 @@ class ServerSession(
     _initialized: InitializationState = InitializationState.NotInitialized
     _client_params: types.InitializeRequestParams | None = None
     _experimental_features: ExperimentalServerSessionFeatures | None = None
+    _tenant_id: str | None = None
 
     def __init__(
         self,
@@ -108,6 +109,16 @@ def _receive_notification_adapter(self) -> TypeAdapter[types.ClientNotification]
     def client_params(self) -> types.InitializeRequestParams | None:
         return self._client_params
 
+    @property
+    def tenant_id(self) -> str | None:
+        """Get the tenant_id for this session."""
+        return self._tenant_id
+
+    @tenant_id.setter
+    def tenant_id(self, value: str | None) -> None:
+        """Set the tenant_id for this session."""
+        self._tenant_id = value
+
     @property
     def experimental(self) -> ExperimentalServerSessionFeatures:
         """Experimental APIs for server→client task operations.
diff --git a/src/mcp/shared/_context.py b/src/mcp/shared/_context.py
index bbcee2d02c..5eeb8dda1b 100644
--- a/src/mcp/shared/_context.py
+++ b/src/mcp/shared/_context.py
@@ -22,3 +22,4 @@ class RequestContext(Generic[SessionT]):
     session: SessionT
     request_id: RequestId | None = None
     meta: RequestParamsMeta | None = None
+    tenant_id: str | None = None
diff --git a/tests/server/auth/middleware/test_auth_context.py b/tests/server/auth/middleware/test_auth_context.py
index 66481bcf79..0aa02eb105 100644
--- a/tests/server/auth/middleware/test_auth_context.py
+++ b/tests/server/auth/middleware/test_auth_context.py
@@ -9,6 +9,7 @@
     AuthContextMiddleware,
     auth_context_var,
     get_access_token,
+    get_tenant_id,
 )
 from mcp.server.auth.middleware.bearer_auth import AuthenticatedUser
 from mcp.server.auth.provider import AccessToken
@@ -117,3 +118,78 @@ async def send(message: Message) -> None:  # pragma: no cover
     # Verify context is still empty after middleware
     assert auth_context_var.get() is None
     assert get_access_token() is None
+
+
+@pytest.fixture
+def access_token_with_tenant() -> AccessToken:
+    """Create an access token with a tenant_id."""
+    return AccessToken(
+        token="tenant_token",
+        client_id="test_client",
+        scopes=["read", "write"],
+        expires_at=int(time.time()) + 3600,
+        tenant_id="tenant-abc",
+    )
+
+
+def test_get_tenant_id_without_auth_context():
+    """Test get_tenant_id returns None when no auth context exists."""
+    assert auth_context_var.get() is None
+    assert get_tenant_id() is None
+
+
+@pytest.mark.anyio
+async def test_get_tenant_id_with_tenant(access_token_with_tenant: AccessToken):
+    """Test get_tenant_id returns tenant_id when auth context has a tenant."""
+    app = MockApp()
+    middleware = AuthContextMiddleware(app)
+
+    user = AuthenticatedUser(access_token_with_tenant)
+    scope: Scope = {"type": "http", "user": user}
+
+    tenant_id_during_call: str | None = None
+
+    class TenantCheckApp:
+        async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
+            nonlocal tenant_id_during_call
+            tenant_id_during_call = get_tenant_id()
+
+    middleware = AuthContextMiddleware(TenantCheckApp())
+
+    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 tenant_id_during_call == "tenant-abc"
+    # Verify context is reset after middleware
+    assert get_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."""
+    tenant_id_during_call: str | None = "not-none"
+
+    class TenantCheckApp:
+        async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
+            nonlocal tenant_id_during_call
+            tenant_id_during_call = get_tenant_id()
+
+    middleware = AuthContextMiddleware(TenantCheckApp())
+
+    user = AuthenticatedUser(valid_access_token)
+    scope: Scope = {"type": "http", "user": user}
+
+    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 tenant_id_during_call is None
diff --git a/tests/server/test_multi_tenancy_session.py b/tests/server/test_multi_tenancy_session.py
new file mode 100644
index 0000000000..b412544dc3
--- /dev/null
+++ b/tests/server/test_multi_tenancy_session.py
@@ -0,0 +1,163 @@
+"""Tests for multi-tenancy support in session and request context."""
+
+import time
+
+import anyio
+import pytest
+
+from mcp.server.auth.middleware.auth_context import auth_context_var, get_tenant_id
+from mcp.server.auth.middleware.bearer_auth import AuthenticatedUser
+from mcp.server.auth.provider import AccessToken
+from mcp.server.context import ServerRequestContext
+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.message import SessionMessage
+from mcp.shared.session import BaseSession
+from mcp.types import ServerCapabilities
+
+
+@pytest.fixture
+def init_options() -> InitializationOptions:
+    """Create initialization options for testing."""
+    return InitializationOptions(
+        server_name="test-server",
+        server_version="1.0.0",
+        capabilities=ServerCapabilities(),
+    )
+
+
+def test_request_context_with_tenant_id():
+    """Test RequestContext can hold tenant_id."""
+    # Use type: ignore since we're testing the dataclass field, not session behavior
+    ctx: RequestContext[BaseSession] = RequestContext(  # type: ignore[type-arg]
+        session=None,  # type: ignore[arg-type]
+        request_id="test-1",
+        tenant_id="tenant-xyz",
+    )
+    assert ctx.tenant_id == "tenant-xyz"
+
+
+def test_request_context_without_tenant_id():
+    """Test RequestContext defaults tenant_id to None."""
+    ctx: RequestContext[BaseSession] = RequestContext(  # type: ignore[type-arg]
+        session=None,  # type: ignore[arg-type]
+        request_id="test-1",
+    )
+    assert ctx.tenant_id is None
+
+
+def test_server_request_context_with_tenant_id():
+    """Test ServerRequestContext can hold tenant_id."""
+    ctx = ServerRequestContext(
+        session=None,  # type: ignore[arg-type]
+        lifespan_context={},
+        experimental=Experimental(
+            task_metadata=None,
+            _client_capabilities=None,
+            _session=None,  # type: ignore[arg-type]
+            _task_support=None,
+        ),
+        tenant_id="tenant-abc",
+    )
+    assert ctx.tenant_id == "tenant-abc"
+
+
+def test_server_request_context_inherits_tenant_id_from_base():
+    """Test ServerRequestContext inherits tenant_id behavior from RequestContext."""
+    # Without tenant_id
+    ctx_no_tenant = ServerRequestContext(
+        session=None,  # type: ignore[arg-type]
+        lifespan_context={},
+        experimental=Experimental(
+            task_metadata=None,
+            _client_capabilities=None,
+            _session=None,  # type: ignore[arg-type]
+            _task_support=None,
+        ),
+    )
+    assert ctx_no_tenant.tenant_id is None
+
+    # With tenant_id
+    ctx_with_tenant = ServerRequestContext(
+        session=None,  # type: ignore[arg-type]
+        lifespan_context={},
+        experimental=Experimental(
+            task_metadata=None,
+            _client_capabilities=None,
+            _session=None,  # type: ignore[arg-type]
+            _task_support=None,
+        ),
+        tenant_id="my-tenant",
+    )
+    assert ctx_with_tenant.tenant_id == "my-tenant"
+
+
+@pytest.mark.anyio
+async def test_server_session_tenant_id_property(init_options: InitializationOptions):
+    """Test ServerSession tenant_id property and setter."""
+    write_send_stream, write_recv_stream = anyio.create_memory_object_stream[SessionMessage](1)
+    read_send_stream, read_recv_stream = anyio.create_memory_object_stream[SessionMessage | Exception](1)
+
+    session = ServerSession(
+        read_stream=read_recv_stream,
+        write_stream=write_send_stream,
+        init_options=init_options,
+    )
+
+    # Default tenant_id is None
+    assert session.tenant_id is None
+
+    # Can set tenant_id
+    session.tenant_id = "tenant-123"
+    assert session.tenant_id == "tenant-123"
+
+    # Can change tenant_id
+    session.tenant_id = "tenant-456"
+    assert session.tenant_id == "tenant-456"
+
+    # Can reset to None
+    session.tenant_id = None
+    assert session.tenant_id is None
+
+    # Clean up all streams
+    await write_send_stream.aclose()
+    await write_recv_stream.aclose()
+    await read_send_stream.aclose()
+    await read_recv_stream.aclose()
+
+
+def test_get_tenant_id_from_auth_context():
+    """Test get_tenant_id extracts tenant_id from auth context."""
+    # No auth context
+    assert get_tenant_id() is None
+
+    # With auth context but no tenant
+    access_token_no_tenant = AccessToken(
+        token="token1",
+        client_id="client1",
+        scopes=["read"],
+        expires_at=int(time.time()) + 3600,
+    )
+    user_no_tenant = AuthenticatedUser(access_token_no_tenant)
+    token = auth_context_var.set(user_no_tenant)
+    try:
+        assert get_tenant_id() is None
+    finally:
+        auth_context_var.reset(token)
+
+    # With auth context and tenant
+    access_token_with_tenant = AccessToken(
+        token="token2",
+        client_id="client2",
+        scopes=["read"],
+        expires_at=int(time.time()) + 3600,
+        tenant_id="tenant-xyz",
+    )
+    user_with_tenant = AuthenticatedUser(access_token_with_tenant)
+    token = auth_context_var.set(user_with_tenant)
+    try:
+        assert get_tenant_id() == "tenant-xyz"
+    finally:
+        auth_context_var.reset(token)

From f597175b0f5c572e3cc5305d0b71a19f22bcf65e Mon Sep 17 00:00:00 2001
From: Andy Lim <aclim@cisco.com>
Date: Wed, 11 Mar 2026 11:54:29 -0400
Subject: [PATCH 02/14] fix(test): use async context manager for ServerSession
 test

Use proper async with pattern to ensure ServerSession's internal
streams are cleaned up correctly, preventing resource warnings.
---
 tests/server/test_multi_tenancy_session.py | 52 ++++++++++------------
 1 file changed, 23 insertions(+), 29 deletions(-)

diff --git a/tests/server/test_multi_tenancy_session.py b/tests/server/test_multi_tenancy_session.py
index b412544dc3..433b2cb709 100644
--- a/tests/server/test_multi_tenancy_session.py
+++ b/tests/server/test_multi_tenancy_session.py
@@ -97,35 +97,29 @@ def test_server_request_context_inherits_tenant_id_from_base():
 @pytest.mark.anyio
 async def test_server_session_tenant_id_property(init_options: InitializationOptions):
     """Test ServerSession tenant_id property and setter."""
-    write_send_stream, write_recv_stream = anyio.create_memory_object_stream[SessionMessage](1)
-    read_send_stream, read_recv_stream = anyio.create_memory_object_stream[SessionMessage | Exception](1)
-
-    session = ServerSession(
-        read_stream=read_recv_stream,
-        write_stream=write_send_stream,
-        init_options=init_options,
-    )
-
-    # Default tenant_id is None
-    assert session.tenant_id is None
-
-    # Can set tenant_id
-    session.tenant_id = "tenant-123"
-    assert session.tenant_id == "tenant-123"
-
-    # Can change tenant_id
-    session.tenant_id = "tenant-456"
-    assert session.tenant_id == "tenant-456"
-
-    # Can reset to None
-    session.tenant_id = None
-    assert session.tenant_id is None
-
-    # Clean up all streams
-    await write_send_stream.aclose()
-    await write_recv_stream.aclose()
-    await read_send_stream.aclose()
-    await read_recv_stream.aclose()
+    server_to_client_send, server_to_client_recv = anyio.create_memory_object_stream[SessionMessage](1)
+    client_to_server_send, client_to_server_recv = anyio.create_memory_object_stream[SessionMessage | Exception](1)
+
+    async with server_to_client_send, server_to_client_recv, client_to_server_send, client_to_server_recv:
+        async with ServerSession(
+            client_to_server_recv,
+            server_to_client_send,
+            init_options,
+        ) as session:
+            # Default tenant_id is None
+            assert session.tenant_id is None
+
+            # Can set tenant_id
+            session.tenant_id = "tenant-123"
+            assert session.tenant_id == "tenant-123"
+
+            # Can change tenant_id
+            session.tenant_id = "tenant-456"
+            assert session.tenant_id == "tenant-456"
+
+            # Can reset to None
+            session.tenant_id = None
+            assert session.tenant_id is None
 
 
 def test_get_tenant_id_from_auth_context():

From a7d9a9f2733e564affc8c25445a2455cacc57a8f Mon Sep 17 00:00:00 2001
From: Andy Lim <aclim@cisco.com>
Date: Wed, 11 Mar 2026 16:23:53 -0400
Subject: [PATCH 03/14] test(auth): add tenant isolation tests for concurrent
 requests

Add two tests to verify tenant_id doesn't leak between:
- Concurrent async requests using the auth contextvar
- Separate ServerSession instances

These tests validate critical security properties for multi-tenant
deployments where isolation between tenants must be guaranteed.
---
 tests/server/test_multi_tenancy_session.py | 116 +++++++++++++++++++++
 1 file changed, 116 insertions(+)

diff --git a/tests/server/test_multi_tenancy_session.py b/tests/server/test_multi_tenancy_session.py
index 433b2cb709..d7acaed7df 100644
--- a/tests/server/test_multi_tenancy_session.py
+++ b/tests/server/test_multi_tenancy_session.py
@@ -155,3 +155,119 @@ def test_get_tenant_id_from_auth_context():
         assert get_tenant_id() == "tenant-xyz"
     finally:
         auth_context_var.reset(token)
+
+
+@pytest.mark.anyio
+async def test_tenant_context_isolation_between_concurrent_requests():
+    """Verify tenant_id doesn't leak between concurrent async contexts.
+
+    This test validates a critical security property: when multiple requests
+    from different tenants are processed concurrently, each request must only
+    see its own tenant_id, never another tenant's.
+
+    How it works:
+    1. We simulate two concurrent requests, each with a different tenant_id
+       ("tenant-A" and "tenant-B").
+
+    2. Each simulated request:
+       - Creates an AccessToken with its tenant_id
+       - Sets it in the auth_context_var (the contextvar used for auth state)
+       - Yields control via anyio.sleep() to allow the other task to run
+       - Reads back the tenant_id via get_tenant_id()
+       - Stores the result for verification
+
+    3. The anyio.sleep(0.01) is intentional - it forces a context switch,
+       creating an opportunity for tenant context to "leak" if the isolation
+       is broken. Without proper contextvar isolation, task2 might see
+       task1's tenant_id (or vice versa) after the context switch.
+
+    4. We use anyio.create_task_group() to run both tasks truly concurrently,
+       not sequentially. This is essential for testing isolation.
+
+    5. Finally, we verify each request saw only its own tenant_id.
+
+    If this test fails, it indicates a serious security issue where tenant
+    data could leak between concurrent requests.
+    """
+    # Store results from each simulated request
+    results: dict[str, str | None] = {}
+
+    async def simulate_request(tenant_id: str, request_key: str) -> None:
+        """Simulate a request with a specific tenant context.
+
+        Args:
+            tenant_id: The tenant_id to set in the auth context
+            request_key: A key to identify this request's result
+        """
+        # Create an access token with the tenant_id, simulating what
+        # the auth middleware does when a request comes in
+        access_token = AccessToken(
+            token=f"token-{request_key}",
+            client_id="test-client",
+            scopes=["read"],
+            expires_at=int(time.time()) + 3600,
+            tenant_id=tenant_id,
+        )
+        user = AuthenticatedUser(access_token)
+
+        # Set the auth context - this is what AuthContextMiddleware does
+        context_token = auth_context_var.set(user)
+        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()
+        finally:
+            # Always reset the context (mirrors middleware behavior)
+            auth_context_var.reset(context_token)
+
+    # Run both requests concurrently using a task group
+    async with anyio.create_task_group() as tg:
+        tg.start_soon(simulate_request, "tenant-A", "request1")
+        tg.start_soon(simulate_request, "tenant-B", "request2")
+
+    # Verify isolation: each request should see only its own tenant_id
+    assert results["request1"] == "tenant-A", "Request 1 saw wrong tenant_id"
+    assert results["request2"] == "tenant-B", "Request 2 saw wrong tenant_id"
+
+
+@pytest.mark.anyio
+async def test_server_session_isolation_between_instances(init_options: InitializationOptions):
+    """Verify tenant_id is isolated between separate ServerSession instances.
+
+    This test ensures that setting tenant_id on one ServerSession does not
+    affect another ServerSession instance. Each session should maintain its
+    own independent tenant context.
+
+    This is important for scenarios where a server handles multiple sessions
+    concurrently - each session belongs to a specific tenant and must not
+    see or affect other tenants' sessions.
+    """
+    # Create streams for two independent sessions
+    send1, recv1 = anyio.create_memory_object_stream[SessionMessage](1)
+    send2, recv2 = anyio.create_memory_object_stream[SessionMessage | Exception](1)
+    send3, recv3 = anyio.create_memory_object_stream[SessionMessage](1)
+    send4, recv4 = anyio.create_memory_object_stream[SessionMessage | Exception](1)
+
+    async with send1, recv1, send2, recv2, send3, recv3, send4, recv4:
+        # Create two separate server sessions
+        async with (
+            ServerSession(recv2, send1, init_options) as session1,
+            ServerSession(recv4, send3, init_options) as session2,
+        ):
+            # Set different tenant_ids on each session
+            session1.tenant_id = "tenant-alpha"
+            session2.tenant_id = "tenant-beta"
+
+            # Verify each session maintains its own tenant_id
+            assert session1.tenant_id == "tenant-alpha"
+            assert session2.tenant_id == "tenant-beta"
+
+            # Modify one session's tenant_id
+            session1.tenant_id = "tenant-gamma"
+
+            # Verify the other session is unaffected
+            assert session1.tenant_id == "tenant-gamma"
+            assert session2.tenant_id == "tenant-beta"  # Still beta, not gamma

From 9f4b679ec7b5a7770bbb346ad4f3418fae20adc4 Mon Sep 17 00:00:00 2001
From: Andy Lim <aclim@cisco.com>
Date: Thu, 12 Mar 2026 08:40:54 -0400
Subject: [PATCH 04/14] docs(context): add tenant_id field description to
 RequestContext docstring

Document the purpose and usage of the tenant_id field for multi-tenant
server deployments.
---
 src/mcp/shared/_context.py | 4 ++++
 1 file changed, 4 insertions(+)

diff --git a/src/mcp/shared/_context.py b/src/mcp/shared/_context.py
index 5eeb8dda1b..88e5d40475 100644
--- a/src/mcp/shared/_context.py
+++ b/src/mcp/shared/_context.py
@@ -17,6 +17,10 @@ class RequestContext(Generic[SessionT]):
 
     For request handlers, request_id is always populated.
     For notification handlers, request_id is None.
+
+    The tenant_id field is used in multi-tenant server deployments to identify
+    which tenant the request belongs to. It is populated from session context
+    and enables tenant-specific request handling and isolation.
     """
 
     session: SessionT

From 9e3ded2b42731076f640413dde8ae20bab3d3a74 Mon Sep 17 00:00:00 2001
From: Andy Lim <aclim@cisco.com>
Date: Thu, 12 Mar 2026 09:19:39 -0400
Subject: [PATCH 05/14] feat(auth): populate session.tenant_id from auth
 context on first request

Wire up session.tenant_id so it is set automatically from the auth
contextvar on the first authenticated request (set-once semantics).
This connects RequestContext.tenant_id and ServerSession.tenant_id,
ensuring the session is bound to a tenant for its lifetime.
---
 src/mcp/server/lowlevel/server.py          | 10 +++-
 tests/server/test_multi_tenancy_session.py | 60 ++++++++++++++++++++++
 2 files changed, 68 insertions(+), 2 deletions(-)

diff --git a/src/mcp/server/lowlevel/server.py b/src/mcp/server/lowlevel/server.py
index 67461465b6..be2c05dbd3 100644
--- a/src/mcp/server/lowlevel/server.py
+++ b/src/mcp/server/lowlevel/server.py
@@ -451,12 +451,15 @@ 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()
+                if tenant_id is not None and session.tenant_id is None:
+                    session.tenant_id = tenant_id
                 ctx = ServerRequestContext(
                     request_id=message.request_id,
                     meta=message.request_meta,
                     session=session,
                     lifespan_context=lifespan_context,
-                    tenant_id=get_tenant_id(),
+                    tenant_id=tenant_id,
                     experimental=Experimental(
                         task_metadata=task_metadata,
                         _client_capabilities=client_capabilities,
@@ -496,10 +499,13 @@ 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()
+                if tenant_id is not None and session.tenant_id is None:
+                    session.tenant_id = tenant_id
                 ctx = ServerRequestContext(
                     session=session,
                     lifespan_context=lifespan_context,
-                    tenant_id=get_tenant_id(),
+                    tenant_id=tenant_id,
                     experimental=Experimental(
                         task_metadata=None,
                         _client_capabilities=client_capabilities,
diff --git a/tests/server/test_multi_tenancy_session.py b/tests/server/test_multi_tenancy_session.py
index d7acaed7df..82dc2a850d 100644
--- a/tests/server/test_multi_tenancy_session.py
+++ b/tests/server/test_multi_tenancy_session.py
@@ -157,6 +157,66 @@ def test_get_tenant_id_from_auth_context():
         auth_context_var.reset(token)
 
 
+@pytest.mark.anyio
+async def test_session_tenant_id_set_from_auth_context_on_first_request(init_options: InitializationOptions):
+    """Verify session.tenant_id is populated from auth context on the first request.
+
+    The lowlevel server sets session.tenant_id from get_tenant_id() on the
+    first request that has a tenant. This test simulates that behavior directly.
+    """
+    server_to_client_send, server_to_client_recv = anyio.create_memory_object_stream[SessionMessage](1)
+    client_to_server_send, client_to_server_recv = anyio.create_memory_object_stream[SessionMessage | Exception](1)
+
+    async with server_to_client_send, server_to_client_recv, client_to_server_send, client_to_server_recv:
+        async with ServerSession(
+            client_to_server_recv,
+            server_to_client_send,
+            init_options,
+        ) as session:
+            assert session.tenant_id is None
+
+            # Simulate what lowlevel/server.py does: set session.tenant_id
+            # from auth context on first request
+            access_token = AccessToken(
+                token="token-first",
+                client_id="client",
+                scopes=["read"],
+                expires_at=int(time.time()) + 3600,
+                tenant_id="tenant-first",
+            )
+            user = AuthenticatedUser(access_token)
+            context_token = auth_context_var.set(user)
+            try:
+                tenant_id = get_tenant_id()
+                if tenant_id is not None and session.tenant_id is None:
+                    session.tenant_id = tenant_id
+            finally:
+                auth_context_var.reset(context_token)
+
+            assert session.tenant_id == "tenant-first"
+
+            # Simulate a second request with a different tenant —
+            # session.tenant_id should NOT change (set-once on first request)
+            access_token2 = AccessToken(
+                token="token-second",
+                client_id="client",
+                scopes=["read"],
+                expires_at=int(time.time()) + 3600,
+                tenant_id="tenant-second",
+            )
+            user2 = AuthenticatedUser(access_token2)
+            context_token2 = auth_context_var.set(user2)
+            try:
+                tenant_id = get_tenant_id()
+                if tenant_id is not None and session.tenant_id is None:
+                    session.tenant_id = tenant_id
+            finally:
+                auth_context_var.reset(context_token2)
+
+            # Still the first tenant — not overwritten
+            assert session.tenant_id == "tenant-first"
+
+
 @pytest.mark.anyio
 async def test_tenant_context_isolation_between_concurrent_requests():
     """Verify tenant_id doesn't leak between concurrent async contexts.

From 161f123ba8b0ed5a83759ad003535ba361066c1d Mon Sep 17 00:00:00 2001
From: Andy Lim <aclim@cisco.com>
Date: Thu, 12 Mar 2026 09:27:18 -0400
Subject: [PATCH 06/14] fix(test): resolve pyright reportUnnecessaryComparison
 error

Extract _simulate_tenant_binding helper to avoid pyright narrowing
session.tenant_id to a literal type after assertion, which caused the
subsequent `is None` check to be flagged as always-False.
---
 tests/server/test_multi_tenancy_session.py | 52 +++++++++-------------
 1 file changed, 21 insertions(+), 31 deletions(-)

diff --git a/tests/server/test_multi_tenancy_session.py b/tests/server/test_multi_tenancy_session.py
index 82dc2a850d..a2c5f91016 100644
--- a/tests/server/test_multi_tenancy_session.py
+++ b/tests/server/test_multi_tenancy_session.py
@@ -18,6 +18,25 @@
 from mcp.types import ServerCapabilities
 
 
+def _simulate_tenant_binding(session: ServerSession, tenant_id_value: str) -> None:
+    """Simulate the set-once tenant binding logic from lowlevel/server.py."""
+    access_token = AccessToken(
+        token=f"token-{tenant_id_value}",
+        client_id="client",
+        scopes=["read"],
+        expires_at=int(time.time()) + 3600,
+        tenant_id=tenant_id_value,
+    )
+    user = AuthenticatedUser(access_token)
+    context_token = auth_context_var.set(user)
+    try:
+        tenant_id = get_tenant_id()
+        if tenant_id is not None and session.tenant_id is None:
+            session.tenant_id = tenant_id
+    finally:
+        auth_context_var.reset(context_token)
+
+
 @pytest.fixture
 def init_options() -> InitializationOptions:
     """Create initialization options for testing."""
@@ -177,41 +196,12 @@ async def test_session_tenant_id_set_from_auth_context_on_first_request(init_opt
 
             # Simulate what lowlevel/server.py does: set session.tenant_id
             # from auth context on first request
-            access_token = AccessToken(
-                token="token-first",
-                client_id="client",
-                scopes=["read"],
-                expires_at=int(time.time()) + 3600,
-                tenant_id="tenant-first",
-            )
-            user = AuthenticatedUser(access_token)
-            context_token = auth_context_var.set(user)
-            try:
-                tenant_id = get_tenant_id()
-                if tenant_id is not None and session.tenant_id is None:
-                    session.tenant_id = tenant_id
-            finally:
-                auth_context_var.reset(context_token)
-
+            _simulate_tenant_binding(session, "tenant-first")
             assert session.tenant_id == "tenant-first"
 
             # Simulate a second request with a different tenant —
             # session.tenant_id should NOT change (set-once on first request)
-            access_token2 = AccessToken(
-                token="token-second",
-                client_id="client",
-                scopes=["read"],
-                expires_at=int(time.time()) + 3600,
-                tenant_id="tenant-second",
-            )
-            user2 = AuthenticatedUser(access_token2)
-            context_token2 = auth_context_var.set(user2)
-            try:
-                tenant_id = get_tenant_id()
-                if tenant_id is not None and session.tenant_id is None:
-                    session.tenant_id = tenant_id
-            finally:
-                auth_context_var.reset(context_token2)
+            _simulate_tenant_binding(session, "tenant-second")
 
             # Still the first tenant — not overwritten
             assert session.tenant_id == "tenant-first"

From f3ed0990107d97511f3eb7af98421d937dcf6086 Mon Sep 17 00:00:00 2001
From: Andy Lim <aclim@cisco.com>
Date: Thu, 12 Mar 2026 09:41:11 -0400
Subject: [PATCH 07/14] test(auth): add E2E tests for tenant_id binding in
 request/notification handling

Add two E2E tests using Client(server) that exercise the session.tenant_id
set-once binding in lowlevel/server.py, covering the previously uncovered
branches in _handle_request (line 456) and _handle_notification (line 504).
---
 tests/server/test_multi_tenancy_session.py | 86 +++++++++++++++++++++-
 1 file changed, 85 insertions(+), 1 deletion(-)

diff --git a/tests/server/test_multi_tenancy_session.py b/tests/server/test_multi_tenancy_session.py
index a2c5f91016..870607bbb6 100644
--- a/tests/server/test_multi_tenancy_session.py
+++ b/tests/server/test_multi_tenancy_session.py
@@ -5,6 +5,8 @@
 import anyio
 import pytest
 
+from mcp import Client
+from mcp.server import Server
 from mcp.server.auth.middleware.auth_context import auth_context_var, get_tenant_id
 from mcp.server.auth.middleware.bearer_auth import AuthenticatedUser
 from mcp.server.auth.provider import AccessToken
@@ -15,7 +17,7 @@
 from mcp.shared._context import RequestContext
 from mcp.shared.message import SessionMessage
 from mcp.shared.session import BaseSession
-from mcp.types import ServerCapabilities
+from mcp.types import ListToolsResult, NotificationParams, PaginatedRequestParams, ServerCapabilities
 
 
 def _simulate_tenant_binding(session: ServerSession, tenant_id_value: str) -> None:
@@ -321,3 +323,85 @@ async def test_server_session_isolation_between_instances(init_options: Initiali
             # Verify the other session is unaffected
             assert session1.tenant_id == "tenant-gamma"
             assert session2.tenant_id == "tenant-beta"  # Still beta, not gamma
+
+
+@pytest.mark.anyio
+async def test_handle_request_populates_session_tenant_id():
+    """E2E: session.tenant_id is set from auth context during request handling.
+
+    This exercises the set-once tenant binding in lowlevel/server.py
+    _handle_request, covering the branch where get_tenant_id() returns
+    a non-None value.
+    """
+    captured_ctx_tenant: str | None = None
+    captured_session_tenant: str | None = None
+
+    async def handle_list_tools(
+        ctx: ServerRequestContext, params: PaginatedRequestParams | None
+    ) -> ListToolsResult:
+        nonlocal captured_ctx_tenant, captured_session_tenant
+        captured_ctx_tenant = ctx.tenant_id
+        captured_session_tenant = ctx.session.tenant_id
+        return ListToolsResult(tools=[])
+
+    server = Server("test", on_list_tools=handle_list_tools)
+
+    # Set auth context with tenant before entering the Client —
+    # contextvars are inherited by child tasks, so the server will see it
+    access_token = AccessToken(
+        token="test-token",
+        client_id="test-client",
+        scopes=["read"],
+        expires_at=int(time.time()) + 3600,
+        tenant_id="tenant-e2e",
+    )
+    user = AuthenticatedUser(access_token)
+    token = auth_context_var.set(user)
+    try:
+        async with Client(server) as client:
+            await client.list_tools()
+    finally:
+        auth_context_var.reset(token)
+
+    assert captured_ctx_tenant == "tenant-e2e"
+    assert captured_session_tenant == "tenant-e2e"
+
+
+@pytest.mark.anyio
+async def test_handle_notification_populates_session_tenant_id():
+    """E2E: session.tenant_id is set from auth context during notification handling.
+
+    This exercises the set-once tenant binding in lowlevel/server.py
+    _handle_notification, covering the branch where get_tenant_id() returns
+    a non-None value.
+    """
+    notification_tenant: str | None = None
+    notification_received = anyio.Event()
+
+    async def handle_roots_list_changed(
+        ctx: ServerRequestContext, params: NotificationParams | None
+    ) -> None:
+        nonlocal notification_tenant
+        notification_tenant = ctx.tenant_id
+        notification_received.set()
+
+    server = Server("test", on_roots_list_changed=handle_roots_list_changed)
+
+    access_token = AccessToken(
+        token="test-token",
+        client_id="test-client",
+        scopes=["read"],
+        expires_at=int(time.time()) + 3600,
+        tenant_id="tenant-notify",
+    )
+    user = AuthenticatedUser(access_token)
+    token = auth_context_var.set(user)
+    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)
+
+    assert notification_tenant == "tenant-notify"

From 2dcebd03987edee35130fb124ee991a1ea8baa13 Mon Sep 17 00:00:00 2001
From: Andy Lim <aclim@cisco.com>
Date: Thu, 12 Mar 2026 09:46:52 -0400
Subject: [PATCH 08/14] style: apply ruff formatting to test file

---
 tests/server/test_multi_tenancy_session.py | 8 ++------
 1 file changed, 2 insertions(+), 6 deletions(-)

diff --git a/tests/server/test_multi_tenancy_session.py b/tests/server/test_multi_tenancy_session.py
index 870607bbb6..166dc3a176 100644
--- a/tests/server/test_multi_tenancy_session.py
+++ b/tests/server/test_multi_tenancy_session.py
@@ -336,9 +336,7 @@ async def test_handle_request_populates_session_tenant_id():
     captured_ctx_tenant: str | None = None
     captured_session_tenant: str | None = None
 
-    async def handle_list_tools(
-        ctx: ServerRequestContext, params: PaginatedRequestParams | None
-    ) -> ListToolsResult:
+    async def handle_list_tools(ctx: ServerRequestContext, params: PaginatedRequestParams | None) -> ListToolsResult:
         nonlocal captured_ctx_tenant, captured_session_tenant
         captured_ctx_tenant = ctx.tenant_id
         captured_session_tenant = ctx.session.tenant_id
@@ -378,9 +376,7 @@ async def test_handle_notification_populates_session_tenant_id():
     notification_tenant: str | None = None
     notification_received = anyio.Event()
 
-    async def handle_roots_list_changed(
-        ctx: ServerRequestContext, params: NotificationParams | None
-    ) -> None:
+    async def handle_roots_list_changed(ctx: ServerRequestContext, params: NotificationParams | None) -> None:
         nonlocal notification_tenant
         notification_tenant = ctx.tenant_id
         notification_received.set()

From ec564e71ac558f6540d852e60a5e5395d995489a Mon Sep 17 00:00:00 2001
From: Andy Lim <aclim@cisco.com>
Date: Thu, 12 Mar 2026 09:49:56 -0400
Subject: [PATCH 09/14] fix: remove stale pragma no cover from
 send_roots_list_changed

This line is now covered by the E2E tenant notification test.
---
 src/mcp/client/session.py | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/src/mcp/client/session.py b/src/mcp/client/session.py
index a0ca751bd7..63459c2d43 100644
--- a/src/mcp/client/session.py
+++ b/src/mcp/client/session.py
@@ -407,7 +407,7 @@ async def list_tools(self, *, params: types.PaginatedRequestParams | None = None
 
         return result
 
-    async def send_roots_list_changed(self) -> None:  # pragma: no cover
+    async def send_roots_list_changed(self) -> None:
         """Send a roots/list_changed notification."""
         await self.send_notification(types.RootsListChangedNotification())
 

From 3dfb2d7743fe3e48ac171c14f6005dc476eea24d Mon Sep 17 00:00:00 2001
From: Andy Lim <aclim@cisco.com>
Date: Thu, 12 Mar 2026 11:17:53 -0400
Subject: [PATCH 10/14] feat(auth): enforce set-once semantics on
 ServerSession.tenant_id

Make the tenant_id setter raise ValueError if attempting to change
to a different value once already set. This prevents accidental tenant
reassignment which could be a security issue. Setting to the same
value is allowed (idempotent).
---
 src/mcp/server/session.py                  | 13 ++++++++-
 tests/server/test_multi_tenancy_session.py | 32 ++++++++++++++--------
 2 files changed, 32 insertions(+), 13 deletions(-)

diff --git a/src/mcp/server/session.py b/src/mcp/server/session.py
index fde9836b84..d36e88343a 100644
--- a/src/mcp/server/session.py
+++ b/src/mcp/server/session.py
@@ -116,7 +116,18 @@ def tenant_id(self) -> str | None:
 
     @tenant_id.setter
     def tenant_id(self, value: str | None) -> None:
-        """Set the tenant_id for this session."""
+        """Set the tenant_id for this session (set-once).
+
+        Once a session is bound to a tenant, the tenant_id cannot be changed.
+        This prevents accidental tenant reassignment which could be a security issue.
+
+        Raises:
+            ValueError: If tenant_id is already set to a different value.
+        """
+        if self._tenant_id is not None and value != self._tenant_id:
+            raise ValueError(
+                f"Cannot change tenant_id from '{self._tenant_id}' to '{value}': session is already bound to a tenant"
+            )
         self._tenant_id = value
 
     @property
diff --git a/tests/server/test_multi_tenancy_session.py b/tests/server/test_multi_tenancy_session.py
index 166dc3a176..3dece1536f 100644
--- a/tests/server/test_multi_tenancy_session.py
+++ b/tests/server/test_multi_tenancy_session.py
@@ -117,7 +117,7 @@ def test_server_request_context_inherits_tenant_id_from_base():
 
 @pytest.mark.anyio
 async def test_server_session_tenant_id_property(init_options: InitializationOptions):
-    """Test ServerSession tenant_id property and setter."""
+    """Test ServerSession tenant_id property with set-once semantics."""
     server_to_client_send, server_to_client_recv = anyio.create_memory_object_stream[SessionMessage](1)
     client_to_server_send, client_to_server_recv = anyio.create_memory_object_stream[SessionMessage | Exception](1)
 
@@ -134,13 +134,20 @@ async def test_server_session_tenant_id_property(init_options: InitializationOpt
             session.tenant_id = "tenant-123"
             assert session.tenant_id == "tenant-123"
 
-            # Can change tenant_id
-            session.tenant_id = "tenant-456"
-            assert session.tenant_id == "tenant-456"
+            # Setting to the same value is allowed
+            session.tenant_id = "tenant-123"
+            assert session.tenant_id == "tenant-123"
 
-            # Can reset to None
-            session.tenant_id = None
-            assert session.tenant_id is None
+            # Cannot change to a different value
+            with pytest.raises(ValueError, match="Cannot change tenant_id"):
+                session.tenant_id = "tenant-456"
+
+            # Cannot reset to None once set
+            with pytest.raises(ValueError, match="Cannot change tenant_id"):
+                session.tenant_id = None
+
+            # Original value is preserved
+            assert session.tenant_id == "tenant-123"
 
 
 def test_get_tenant_id_from_auth_context():
@@ -317,12 +324,13 @@ async def test_server_session_isolation_between_instances(init_options: Initiali
             assert session1.tenant_id == "tenant-alpha"
             assert session2.tenant_id == "tenant-beta"
 
-            # Modify one session's tenant_id
-            session1.tenant_id = "tenant-gamma"
+            # Attempting to change one session's tenant_id raises
+            with pytest.raises(ValueError, match="Cannot change tenant_id"):
+                session1.tenant_id = "tenant-gamma"
 
-            # Verify the other session is unaffected
-            assert session1.tenant_id == "tenant-gamma"
-            assert session2.tenant_id == "tenant-beta"  # Still beta, not gamma
+            # Both sessions retain their original values
+            assert session1.tenant_id == "tenant-alpha"
+            assert session2.tenant_id == "tenant-beta"
 
 
 @pytest.mark.anyio

From 117f0da02cb1f123624f2feb614f40bb20bbee48 Mon Sep 17 00:00:00 2001
From: Andy Lim <aclim@cisco.com>
Date: Fri, 13 Mar 2026 14:52:52 -0400
Subject: [PATCH 11/14] refactor(auth): decouple core server from auth module
 for tenant extraction
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

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.
---
 .../server/auth/middleware/auth_context.py    |  9 ++-
 src/mcp/server/lowlevel/server.py             |  7 ++-
 src/mcp/shared/_context.py                    |  6 ++
 .../auth/middleware/test_auth_context.py      | 55 +++++++++++++++++++
 tests/server/test_multi_tenancy_session.py    | 38 ++++++++-----
 5 files changed, 97 insertions(+), 18 deletions(-)

diff --git a/src/mcp/server/auth/middleware/auth_context.py b/src/mcp/server/auth/middleware/auth_context.py
index c619d81a56..2cee836e1c 100644
--- a/src/mcp/server/auth/middleware/auth_context.py
+++ b/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)
diff --git a/src/mcp/server/lowlevel/server.py b/src/mcp/server/lowlevel/server.py
index be2c05dbd3..97d39f6de1 100644
--- a/src/mcp/server/lowlevel/server.py
+++ b/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(
diff --git a/src/mcp/shared/_context.py b/src/mcp/shared/_context.py
index 88e5d40475..3b4f5967a5 100644
--- a/src/mcp/shared/_context.py
+++ b/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])
 
 
diff --git a/tests/server/auth/middleware/test_auth_context.py b/tests/server/auth/middleware/test_auth_context.py
index 0aa02eb105..1d50c69029 100644
--- a/tests/server/auth/middleware/test_auth_context.py
+++ b/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."""
diff --git a/tests/server/test_multi_tenancy_session.py b/tests/server/test_multi_tenancy_session.py
index 3dece1536f..4e5e1acd5a 100644
--- a/tests/server/test_multi_tenancy_session.py
+++ b/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"

From 02725c8bb86ad269ce1dd47c9d8e6e01597049b7 Mon Sep 17 00:00:00 2001
From: Andy Lim <aclim@cisco.com>
Date: Mon, 16 Mar 2026 13:59:31 -0400
Subject: [PATCH 12/14] fix(test): remove dead code in
 test_get_tenant_id_with_tenant

Remove unused MockApp() and AuthContextMiddleware(app) that were
immediately overwritten by AuthContextMiddleware(TenantCheckApp()).
---
 tests/server/auth/middleware/test_auth_context.py | 3 ---
 1 file changed, 3 deletions(-)

diff --git a/tests/server/auth/middleware/test_auth_context.py b/tests/server/auth/middleware/test_auth_context.py
index 1d50c69029..74736ef5b8 100644
--- a/tests/server/auth/middleware/test_auth_context.py
+++ b/tests/server/auth/middleware/test_auth_context.py
@@ -142,9 +142,6 @@ def test_get_tenant_id_without_auth_context():
 @pytest.mark.anyio
 async def test_get_tenant_id_with_tenant(access_token_with_tenant: AccessToken):
     """Test get_tenant_id returns tenant_id when auth context has a tenant."""
-    app = MockApp()
-    middleware = AuthContextMiddleware(app)
-
     user = AuthenticatedUser(access_token_with_tenant)
     scope: Scope = {"type": "http", "user": user}
 

From 04c7535c1b7d2843d9851c4b5b5f67dc9c5f34ef Mon Sep 17 00:00:00 2001
From: Andy Lim <aclim@cisco.com>
Date: Mon, 16 Mar 2026 14:04:42 -0400
Subject: [PATCH 13/14] fix(test): replace anyio.sleep(0.01) with
 anyio.lowlevel.checkpoint()

Use checkpoint() for deterministic context switching instead of a
fixed-duration sleep in the tenant isolation test.
---
 tests/server/test_multi_tenancy_session.py | 10 +++++-----
 1 file changed, 5 insertions(+), 5 deletions(-)

diff --git a/tests/server/test_multi_tenancy_session.py b/tests/server/test_multi_tenancy_session.py
index 4e5e1acd5a..b5f99c0f85 100644
--- a/tests/server/test_multi_tenancy_session.py
+++ b/tests/server/test_multi_tenancy_session.py
@@ -237,13 +237,13 @@ async def test_tenant_context_isolation_between_concurrent_requests():
     2. Each simulated request:
        - Creates an AccessToken with its tenant_id
        - Sets it in the auth_context_var (the contextvar used for auth state)
-       - Yields control via anyio.sleep() to allow the other task to run
+       - Yields control via checkpoint() to allow the other task to run
        - Reads back the tenant_id via get_tenant_id()
        - Stores the result for verification
 
-    3. The anyio.sleep(0.01) is intentional - it forces a context switch,
-       creating an opportunity for tenant context to "leak" if the isolation
-       is broken. Without proper contextvar isolation, task2 might see
+    3. The anyio.lowlevel.checkpoint() forces a context switch, creating
+       an opportunity for tenant context to "leak" if the isolation is
+       broken. Without proper contextvar isolation, task2 might see
        task1's tenant_id (or vice versa) after the context switch.
 
     4. We use anyio.create_task_group() to run both tasks truly concurrently,
@@ -281,7 +281,7 @@ async def simulate_request(tenant_id: str, request_key: str) -> None:
         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)
+            await anyio.lowlevel.checkpoint()
 
             # Read back the tenant_id - should still be our tenant, not the other
             results[request_key] = tenant_id_var.get()

From 92dff290b838907688c6bb500f433ad9472f80a0 Mon Sep 17 00:00:00 2001
From: Andy Lim <aclim@cisco.com>
Date: Mon, 16 Mar 2026 14:08:30 -0400
Subject: [PATCH 14/14] fix(test): use explicit import for
 anyio.lowlevel.checkpoint

Import checkpoint directly from anyio.lowlevel to fix pyright
reportAttributeAccessIssue on the lazy submodule.
---
 tests/server/test_multi_tenancy_session.py | 3 ++-
 1 file changed, 2 insertions(+), 1 deletion(-)

diff --git a/tests/server/test_multi_tenancy_session.py b/tests/server/test_multi_tenancy_session.py
index b5f99c0f85..2f94fa06db 100644
--- a/tests/server/test_multi_tenancy_session.py
+++ b/tests/server/test_multi_tenancy_session.py
@@ -4,6 +4,7 @@
 
 import anyio
 import pytest
+from anyio.lowlevel import checkpoint
 
 from mcp import Client
 from mcp.server import Server
@@ -281,7 +282,7 @@ async def simulate_request(tenant_id: str, request_key: str) -> None:
         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.lowlevel.checkpoint()
+            await checkpoint()
 
             # Read back the tenant_id - should still be our tenant, not the other
             results[request_key] = tenant_id_var.get()