feat(auth): add tenant_id to session and request context

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.

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
@@ -456,6 +456,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,
@@ -498,6 +499,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,

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.

src/mcp/shared/_context.py

@@ -17,8 +17,13 @@ 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
     request_id: RequestId | None = None
     meta: RequestParamsMeta | None = None
+    tenant_id: str | None = None

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