Add multi-tenancy TODO comments across MCP Python SDK codebase

src/mcp/client/session.py

@@ -108,6 +108,9 @@ class ClientSession(
         types.ServerNotification,
     ]
 ):
+    # TODO: Multi-tenancy - ClientSession does not track or send tenant_id. Need to add tenant_id
+    # field and pass it to the server during initialization and with each request (either in request
+    # metadata or as a header). This allows the server to process requests in the correct tenant context.
     def __init__(
         self,
         read_stream: MemoryObjectReceiveStream[SessionMessage | Exception],
@@ -137,6 +140,9 @@ def __init__(
         self._server_capabilities: types.ServerCapabilities | None = None
 
     async def initialize(self) -> types.InitializeResult:
+        # TODO: Multi-tenancy - Client initialization does not send tenant_id to server.
+        # Need to add tenant_id parameter to initialize() and include it in InitializeRequestParams
+        # or as part of clientInfo metadata. Server must know which tenant context to use.
         sampling = types.SamplingCapability() if self._sampling_callback is not _default_sampling_callback else None
         elicitation = (
             types.ElicitationCapability() if self._elicitation_callback is not _default_elicitation_callback else None

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

@@ -17,6 +17,9 @@ def get_access_token() -> AccessToken | None:
     Returns:
         The access token if an authenticated user is available, None otherwise.
     """
+    # TODO: Multi-tenancy - Need to add get_tenant_id() helper function that extracts
+    # tenant_id from the AccessToken in the current auth context. All data access operations
+    # should use this tenant_id to scope queries and prevent cross-tenant data access.
     auth_user = auth_context_var.get()
     return auth_user.access_token if auth_user else None
 

src/mcp/server/auth/provider.py

@@ -17,6 +17,9 @@ class AuthorizationParams(BaseModel):
 
 
 class AuthorizationCode(BaseModel):
+    # TODO: Multi-tenancy - AuthorizationCode lacks tenant_id field. Need to add tenant_id
+    # to track which tenant this authorization code belongs to, ensuring codes cannot be
+    # exchanged across tenant boundaries.
     code: str
     scopes: list[str]
     expires_at: float
@@ -28,13 +31,19 @@ class AuthorizationCode(BaseModel):
 
 
 class RefreshToken(BaseModel):
+    # TODO: Multi-tenancy - RefreshToken lacks tenant_id field. Need to add tenant_id
+    # to ensure refresh tokens are scoped to specific tenants and cannot be used to
+    # obtain access tokens for other tenants.
     token: str
     client_id: str
     scopes: list[str]
     expires_at: int | None = None
 
 
 class AccessToken(BaseModel):
+    # TODO: Multi-tenancy - AccessToken lacks tenant_id field. This is critical - need to add
+    # tenant_id to ensure access tokens are scoped to specific tenants. All resource/tool/prompt
+    # access should be filtered by the tenant_id from the access token to prevent cross-tenant access.
     token: str
     client_id: str
     scopes: list[str]

src/mcp/server/fastmcp/prompts/manager.py

@@ -19,6 +19,10 @@ class PromptManager:
     """Manages FastMCP prompts."""
 
     def __init__(self, warn_on_duplicate_prompts: bool = True):
+        # TODO: Multi-tenancy - Prompts are stored in a shared dictionary without tenant scoping.
+        # Need to either: (1) add tenant_id parameter to all methods and scope storage by tenant
+        # (e.g., dict[tuple[tenant_id, prompt_name], Prompt]), or (2) create separate PromptManager
+        # instances per tenant. Prompts registered by one tenant should not be accessible to others.
         self._prompts: dict[str, Prompt] = {}
         self.warn_on_duplicate_prompts = warn_on_duplicate_prompts
 

src/mcp/server/fastmcp/resources/resource_manager.py

@@ -24,6 +24,11 @@ class ResourceManager:
     """Manages FastMCP resources."""
 
     def __init__(self, warn_on_duplicate_resources: bool = True):
+        # TODO: Multi-tenancy - Resources and templates are stored in shared dictionaries
+        # without tenant scoping. Need to either: (1) add tenant_id parameter to all methods
+        # and scope storage by tenant (e.g., dict[tuple[tenant_id, uri], Resource]), or
+        # (2) create separate ResourceManager instances per tenant. Resources registered by
+        # one tenant should not be accessible to other tenants.
         self._resources: dict[str, Resource] = {}
         self._templates: dict[str, ResourceTemplate] = {}
         self.warn_on_duplicate_resources = warn_on_duplicate_resources

src/mcp/server/fastmcp/server.py

@@ -78,6 +78,9 @@ class Settings(BaseSettings, Generic[LifespanResultT]):
     For example, FASTMCP_DEBUG=true will set debug=True.
     """
 
+    # TODO: Multi-tenancy - Settings are loaded from environment variables without tenant scoping.
+    # For multi-tenant deployments, need to support tenant-specific configuration overrides,
+    # potentially loading from tenant-scoped config sources (e.g., database, per-tenant env files).
     model_config = SettingsConfigDict(
         env_prefix="FASTMCP_",
         env_file=".env",
@@ -198,6 +201,10 @@ def __init__(  # noqa: PLR0913
             # We need to create a Lifespan type that is a generic on the server type, like Starlette does.
             lifespan=(lifespan_wrapper(self, self.settings.lifespan) if self.settings.lifespan else default_lifespan),  # type: ignore
         )
+        # TODO: Multi-tenancy - These managers maintain shared state across all tenants.
+        # Need to either: (1) make managers tenant-aware by accepting tenant_id in all operations,
+        # or (2) create separate manager instances per tenant with tenant-scoped storage.
+        # Tools, resources, and prompts registered should be scoped to tenant context.
         self._tool_manager = ToolManager(tools=tools, warn_on_duplicate_tools=self.settings.warn_on_duplicate_tools)
         self._resource_manager = ResourceManager(warn_on_duplicate_resources=self.settings.warn_on_duplicate_resources)
         self._prompt_manager = PromptManager(warn_on_duplicate_prompts=self.settings.warn_on_duplicate_prompts)
@@ -210,15 +217,23 @@ def __init__(  # noqa: PLR0913
         elif auth_server_provider or token_verifier:
             raise ValueError("Cannot specify auth_server_provider or token_verifier without auth settings")
 
+        # TODO: Multi-tenancy - Auth providers and token verifiers are shared across all tenants.
+        # Need to support tenant-specific auth configurations, or at minimum ensure tokens
+        # include tenant_id claims that are validated. AccessToken should include tenant_id field.
         self._auth_server_provider = auth_server_provider
         self._token_verifier = token_verifier
 
         # Create token verifier from provider if needed (backwards compatibility)
         if auth_server_provider and not token_verifier:
             self._token_verifier = ProviderTokenVerifier(auth_server_provider)
+        # TODO: Multi-tenancy - Event store is shared across tenants. Events should be
+        # scoped by tenant_id to prevent cross-tenant data leakage in resumable sessions.
         self._event_store = event_store
         self._custom_starlette_routes: list[Route] = []
         self.dependencies = self.settings.dependencies
+        # TODO: Multi-tenancy - Session manager tracks sessions globally without tenant scoping.
+        # Sessions should be partitioned by tenant_id to isolate tenant data and prevent
+        # cross-tenant session access. Consider tenant_id as part of session key.
         self._session_manager: StreamableHTTPSessionManager | None = None
 
         # Set up MCP protocol handlers

src/mcp/server/fastmcp/tools/tool_manager.py

@@ -25,6 +25,10 @@ def __init__(
         *,
         tools: list[Tool] | None = None,
     ):
+        # TODO: Multi-tenancy - Tools are stored in a shared dictionary without tenant scoping.
+        # Need to either: (1) add tenant_id parameter to all methods and scope storage by tenant
+        # (e.g., dict[tuple[tenant_id, tool_name], Tool]), or (2) create separate ToolManager
+        # instances per tenant. Tools registered by one tenant should not be accessible to others.
         self._tools: dict[str, Tool] = {}
         if tools is not None:
             for tool in tools:

src/mcp/server/lowlevel/server.py

@@ -149,10 +149,15 @@ def __init__(
         self.website_url = website_url
         self.icons = icons
         self.lifespan = lifespan
+        # TODO: Multi-tenancy - Request handlers are shared across all tenants. Handlers should
+        # receive tenant context and scope their operations accordingly. Consider tenant_id as part
+        # of the request context to enable tenant-aware data access.
         self.request_handlers: dict[type, Callable[..., Awaitable[types.ServerResult]]] = {
             types.PingRequest: _ping_handler,
         }
         self.notification_handlers: dict[type, Callable[..., Awaitable[None]]] = {}
+        # TODO: Multi-tenancy - Tool cache is shared across all tenants. Need to scope cache by
+        # tenant_id or maintain separate caches per tenant to prevent cross-tenant data leakage.
         self._tool_cache: dict[str, types.Tool] = {}
         logger.debug("Initializing server %r", name)
 

src/mcp/server/session.py

@@ -77,6 +77,9 @@ class ServerSession(
         types.ClientNotification,
     ]
 ):
+    # TODO: Multi-tenancy - ServerSession does not track tenant_id. Need to add tenant_id field
+    # to identify which tenant this session belongs to. This is critical for isolating tenant data
+    # and ensuring requests are processed in the correct tenant context.
     _initialized: InitializationState = InitializationState.NotInitialized
     _client_params: types.InitializeRequestParams | None = None
 

src/mcp/server/streamable_http_manager.py

@@ -69,6 +69,9 @@ def __init__(
 
         # Session tracking (only used if not stateless)
         self._session_creation_lock = anyio.Lock()
+        # TODO: Multi-tenancy - Server instances are tracked globally by session_id without tenant scoping.
+        # Need to ensure sessions are isolated per tenant. Consider using composite key (tenant_id, session_id)
+        # or partitioning _server_instances by tenant_id to prevent cross-tenant session access.
         self._server_instances: dict[str, StreamableHTTPServerTransport] = {}
 
         # The task group will be set during lifespan

src/mcp/shared/context.py

@@ -13,6 +13,9 @@
 
 @dataclass
 class RequestContext(Generic[SessionT, LifespanContextT, RequestT]):
+    # TODO: Multi-tenancy - RequestContext lacks tenant_id field. Need to add tenant_id to track
+    # which tenant this request belongs to. All handlers should use this tenant_id to scope their
+    # operations and prevent cross-tenant data access.
     request_id: RequestId
     meta: RequestParams.Meta | None
     session: SessionT

tests/server/fastmcp/test_integration.py

@@ -3,6 +3,13 @@
 
 These tests validate the proper functioning of FastMCP features using focused,
 single-feature servers across different transports (SSE and StreamableHTTP).
+
+TODO: Multi-tenancy - Add test coverage for multi-tenant scenarios including:
+  - Multiple tenants with isolated resources, tools, and prompts
+  - Tenant-scoped authentication and authorization
+  - Cross-tenant isolation verification (tenant A cannot access tenant B's data)
+  - Session management with tenant context
+  - Configuration isolation between tenants
 """
 # TODO(Marcelo): The `examples` package is not being imported as package. We need to solve this.
 # pyright: reportUnknownMemberType=false