feat: hosting on single-tenant platforms such Amazon Bedrock AgentCore Runtime

Author: tonyxwz

src/mcp/server/fastmcp/server.py

@@ -84,6 +84,8 @@ class Settings(BaseSettings, Generic[LifespanResultT]):
     json_response: bool
     stateless_http: bool
     """Define if the server should create a new transport per request."""
+    single_tenant: bool
+    """Define if the server should use only one transport for all requests."""
 
     # resource settings
     warn_on_duplicate_resources: bool
@@ -139,6 +141,7 @@ def __init__(
         streamable_http_path: str = "/mcp",
         json_response: bool = False,
         stateless_http: bool = False,
+        single_tenant: bool = False,
         warn_on_duplicate_resources: bool = True,
         warn_on_duplicate_tools: bool = True,
         warn_on_duplicate_prompts: bool = True,
@@ -158,6 +161,7 @@ def __init__(
             streamable_http_path=streamable_http_path,
             json_response=json_response,
             stateless_http=stateless_http,
+            single_tenant=single_tenant,
             warn_on_duplicate_resources=warn_on_duplicate_resources,
             warn_on_duplicate_tools=warn_on_duplicate_tools,
             warn_on_duplicate_prompts=warn_on_duplicate_prompts,
@@ -868,6 +872,7 @@ def streamable_http_app(self) -> Starlette:
                 event_store=self._event_store,
                 json_response=self.settings.json_response,
                 stateless=self.settings.stateless_http,  # Use the stateless setting
+                single_tenant=self.settings.single_tenant,
                 security_settings=self.settings.transport_security,
             )
 

src/mcp/server/streamable_http.py

@@ -669,6 +669,7 @@ async def _handle_unsupported_request(self, request: Request, send: Send) -> Non
 
     async def _validate_request_headers(self, request: Request, send: Send) -> bool:
         if not await self._validate_session(request, send):
+            logger.error("Session not valid")
             return False
         if not await self._validate_protocol_version(request, send):
             return False
@@ -685,6 +686,7 @@ async def _validate_session(self, request: Request, send: Send) -> bool:
 
         # If no session ID provided but required, return error
         if not request_session_id:
+            logger.error("Session not present but required")
             response = self._create_error_response(
                 "Bad Request: Missing session ID",
                 HTTPStatus.BAD_REQUEST,
@@ -698,6 +700,7 @@ async def _validate_session(self, request: Request, send: Send) -> bool:
                 "Not Found: Invalid or expired session ID",
                 HTTPStatus.NOT_FOUND,
             )
+            logger.error(f"Session id does not match, {request_session_id=} vs. {self.mcp_session_id=}")
             await response(request.scope, request.receive, send)
             return False
 

src/mcp/server/streamable_http_manager.py

@@ -51,6 +51,9 @@ class StreamableHTTPSessionManager:
         json_response: Whether to use JSON responses instead of SSE streams
         stateless: If True, creates a completely fresh transport for each request
                    with no session tracking or state persistence between requests.
+        single_tenant: If True, only one transport will be created to process the entire
+                    every request, regardless of the MCP session id. This is useful for
+                    hosting platforms where the MCP server is launched in a single-tenant box.
     """
 
     def __init__(
@@ -59,12 +62,18 @@ def __init__(
         event_store: EventStore | None = None,
         json_response: bool = False,
         stateless: bool = False,
+        single_tenant: bool = False,
         security_settings: TransportSecuritySettings | None = None,
     ):
         self.app = app
         self.event_store = event_store
         self.json_response = json_response
         self.stateless = stateless
+        self.single_tenant = single_tenant
+        if self.stateless and self.single_tenant:
+            # A single-tenant server must be stateful, but stateful server does not
+            # have to be single tenant.
+            raise ValueError("A single-tenant server must stateful.")
         self.security_settings = security_settings
 
         # Session tracking (only used if not stateless)
@@ -209,6 +218,19 @@ async def _handle_stateful_request(
         request = Request(scope, receive)
         request_mcp_session_id = request.headers.get(MCP_SESSION_ID_HEADER)
 
+        if self.single_tenant and self._server_instances:
+            # being single_tenant means that there is only one ASGI server for the entire application.
+            # that server is used to process all the mcp requests because the hosting platform
+            # is already distributing the request to the box where the single-tenant mcp server runs.
+            assert len(self._server_instances) == 1
+            # hosting platforms might exposes a different mcp session ID hence
+            # we take the first key of server instances as the mcp session id
+            request_mcp_session_id = next(iter(self._server_instances.keys()))
+            headers = dict(scope["headers"])
+            # Also need to reset the incoming request mcp session id to this existing session id
+            headers[MCP_SESSION_ID_HEADER.encode("latin-1")] = request_mcp_session_id.encode("latin-1")
+            scope["headers"] = list(headers.items())
+
         # Existing session case
         if request_mcp_session_id is not None and request_mcp_session_id in self._server_instances:
             transport = self._server_instances[request_mcp_session_id]