[v1.x] Expire idle Streamable HTTP sessions by default and cap concurrent sessions

v1.x backport of #3395.

Stateful Streamable HTTP sessions get a lifecycle the server owns: a
session is forgotten and its transport terminated as soon as it ends
(client DELETE, a refused or failed opening request, idle expiry,
crash), sessions with nothing in flight expire after
session_idle_timeout (default 1800 seconds; an open GET stream or a
request being answered holds the session), and a manager admits at most
max_sessions (default 10 000) and answers 503 beyond that. A stateless
transport is terminated even when its request is cancelled. Both
settings are FastMCP(...) keyword arguments and Settings fields.

Differences from #3395:
- the per-session task drives Server.run() rather than serve_loop;
- idle expiry ends a session by discarding it (which closes the
  transport's streams, as a client DELETE does) rather than by
  cancelling Server.run(): on 1.x Server.run() enters the server's
  lifespan per session, and its teardown should complete rather than
  run under cancellation;
- the settings are FastMCP constructor arguments / Settings fields
  rather than run()/app-factory keywords, and
  FastMCP.streamable_http_app() takes no options;
- the manager's JSON-RPC error bodies keep this branch's
  id "server-error" envelope;
- the transport also closes its per-request SSE streams when a response
  ends (the 1.x share of an earlier main change), which the in-process
  tests need;
- tests drive the manager through raw ASGI callables;
- documentation lives in docs/server.md.

Author: maxisbey

docs/server.md

@@ -1254,6 +1254,7 @@ The FastMCP server instance accessible via `ctx.fastmcp` provides access to serv
   - `mount_path`, `sse_path`, `streamable_http_path` - Transport paths
   - `stateless_http` - Whether the server operates in stateless mode
   - `max_request_body_size` - Maximum HTTP request body size in bytes (Streamable HTTP and SSE)
+  - `session_idle_timeout` and `max_sessions` - Streamable HTTP session expiry and session cap
   - And other configuration options
 
 ```python
@@ -1426,6 +1427,9 @@ messages, configure the smallest suitable byte limit:
 mcp = FastMCP("Large messages", max_request_body_size=8 * 1024 * 1024)
 ```
 
+Stateful sessions expire and are capped per process. See
+[Session lifetime and limits](#session-lifetime-and-limits) below.
+
 <!-- snippet-source examples/snippets/servers/streamable_config.py -->
 ```python
 """
@@ -1536,6 +1540,39 @@ The streamable HTTP transport supports:
 - JSON or SSE response formats
 - Better scalability for multi-node deployments
 
+#### Session lifetime and limits
+
+A stateful session does not live forever, and one process does not hold an unlimited number of
+them. Two settings control this. Both are keyword arguments on `FastMCP(...)`. `stateless_http=True`
+keeps no sessions, so neither applies there.
+
+| Setting | Default | What it does | What the client sees | Turn it off |
+|---|---|---|---|---|
+| `session_idle_timeout` | `1800` (30 min) | Closes a session that has had nothing in flight for that long. | `404 Session not found`. It has to `initialize` again. | `None` |
+| `max_sessions` | `10_000` | Refuses to open a session beyond that many. Existing sessions are untouched and nothing is evicted. | `503 Too many open sessions` with JSON-RPC code `-32603`. | `None` |
+
+What counts as "in flight":
+
+- An open `GET` stream. The SDK clients keep one open, so a connected client's session never
+  expires.
+- A request that is still being answered. A tool call that runs longer than the timeout is not
+  interrupted, and the countdown only starts once it finishes.
+- Nothing else. Between requests the clock runs. Any request on the session restarts it,
+  `ping` included. Once a session has expired, nothing revives it.
+
+A client that ends its session with `DELETE` frees it immediately. So does a client whose
+opening request was refused.
+
+```python
+mcp = FastMCP("My server", session_idle_timeout=None, max_sessions=50_000)
+```
+
+Both events show up in the server log. An expiry is `Session <id> idle timeout` at `INFO`. A
+refused open is `Refusing to open a new session: <n> sessions are already open` at `WARNING`.
+
+The limits are per process. With four workers the ceiling is four times `max_sessions`, and each
+worker expires its own sessions.
+
 #### CORS Configuration for Browser-Based Clients
 
 If you'd like your server to be accessible by browser-based MCP clients, you'll need to configure CORS headers. The `Mcp-Session-Id` header must be exposed for browser clients to access it:

src/mcp/server/fastmcp/server.py

@@ -62,7 +62,11 @@
 from mcp.server.sse import SseServerTransport
 from mcp.server.stdio import stdio_server
 from mcp.server.streamable_http import EventStore
-from mcp.server.streamable_http_manager import StreamableHTTPSessionManager
+from mcp.server.streamable_http_manager import (
+    DEFAULT_MAX_SESSIONS,
+    DEFAULT_SESSION_IDLE_TIMEOUT,
+    StreamableHTTPSessionManager,
+)
 from mcp.server.transport_security import DEFAULT_MAX_REQUEST_BODY_SIZE, TransportSecuritySettings
 from mcp.shared.context import LifespanContextT, RequestContext, RequestT
 from mcp.types import Annotations, AnyFunction, ContentBlock, GetPromptResult, Icon, ToolAnnotations
@@ -108,6 +112,10 @@ class Settings(BaseSettings, Generic[LifespanResultT]):
     """Define if the server should create a new transport per request."""
     max_request_body_size: int
     """Maximum request body size in bytes for the Streamable HTTP endpoint and the SSE message endpoint."""
+    session_idle_timeout: float | None
+    """Seconds a stateful session may have no request in flight before it is closed. None disables expiry."""
+    max_sessions: int | None
+    """Maximum number of concurrent stateful sessions. None removes the limit."""
 
     # resource settings
     warn_on_duplicate_resources: bool
@@ -169,6 +177,8 @@ def __init__(  # noqa: PLR0913
         json_response: bool = False,
         stateless_http: bool = False,
         max_request_body_size: int = DEFAULT_MAX_REQUEST_BODY_SIZE,
+        session_idle_timeout: float | None = DEFAULT_SESSION_IDLE_TIMEOUT,
+        max_sessions: int | None = DEFAULT_MAX_SESSIONS,
         warn_on_duplicate_resources: bool = True,
         warn_on_duplicate_tools: bool = True,
         warn_on_duplicate_prompts: bool = True,
@@ -197,6 +207,8 @@ def __init__(  # noqa: PLR0913
             json_response=json_response,
             stateless_http=stateless_http,
             max_request_body_size=max_request_body_size,
+            session_idle_timeout=session_idle_timeout,
+            max_sessions=max_sessions,
             warn_on_duplicate_resources=warn_on_duplicate_resources,
             warn_on_duplicate_tools=warn_on_duplicate_tools,
             warn_on_duplicate_prompts=warn_on_duplicate_prompts,
@@ -966,6 +978,8 @@ def streamable_http_app(self) -> Starlette:
                 stateless=self.settings.stateless_http,  # Use the stateless setting
                 security_settings=self.settings.transport_security,
                 max_request_body_size=self.settings.max_request_body_size,
+                session_idle_timeout=self.settings.session_idle_timeout,
+                max_sessions=self.settings.max_sessions,
             )
 
         # Create the ASGI handler

src/mcp/server/streamable_http.py

@@ -9,6 +9,7 @@
 
 import json
 import logging
+import math
 import re
 from abc import ABC, abstractmethod
 from collections.abc import AsyncGenerator, Awaitable, Callable
@@ -150,6 +151,7 @@ def __init__(
         event_store: EventStore | None = None,
         security_settings: TransportSecuritySettings | None = None,
         retry_interval: int | None = None,
+        idle_timeout: float | None = None,
     ) -> None:
         """
         Initialize a new StreamableHTTP server transport.
@@ -167,12 +169,22 @@ def __init__(
                            retry field. When set, the server will send a retry field in
                            SSE priming events to control client reconnection timing for
                            polling behavior. Only used when event_store is provided.
+            idle_timeout: Seconds the session may go without any request in flight before
+                         `idle_scope` is cancelled. A request being served or an open GET
+                         stream holds the session open; the countdown starts each time the
+                         last in-flight request completes. The host waits on `idle_scope`
+                         (available once `connect()` has been entered) and ends the session
+                         when it fires. Default is None: no `idle_scope`, the session never
+                         expires.
 
         Raises:
-            ValueError: If the session ID contains invalid characters.
+            ValueError: If the session ID contains invalid characters, or if `idle_timeout`
+                        is not a positive, finite number.
         """
         if mcp_session_id is not None and not SESSION_ID_PATTERN.fullmatch(mcp_session_id):
             raise ValueError("Session ID must only contain visible ASCII characters (0x21-0x7E)")
+        if idle_timeout is not None and not (math.isfinite(idle_timeout) and idle_timeout > 0):
+            raise ValueError("idle_timeout must be a positive, finite number of seconds")
 
         self.mcp_session_id = mcp_session_id
         self.is_json_response_enabled = is_json_response_enabled
@@ -188,8 +200,11 @@ def __init__(
         ] = {}
         self._sse_stream_writers: dict[RequestId, MemoryObjectSendStream[SSEEvent]] = {}
         self._terminated = False
-        # Idle timeout cancel scope; managed by the session manager.
+        self._idle_timeout = idle_timeout
+        self._requests_in_flight = 0
         self.idle_scope: anyio.CancelScope | None = None
+        """Created when `connect()` is entered if `idle_timeout` is set; cancelled once no request has been in
+        flight for `idle_timeout` seconds."""
 
     @property
     def is_terminated(self) -> bool:
@@ -402,6 +417,32 @@ async def _clean_up_memory_streams(self, request_id: RequestId) -> None:  # prag
 
     async def handle_request(self, scope: Scope, receive: Receive, send: Send) -> None:
         """Application entry point that handles all HTTP requests"""
+        if self.idle_scope is None or self._idle_timeout is None:
+            await self._handle_request(scope, receive, send)
+            return
+
+        if self.idle_scope.cancel_called:
+            # The idle period already ran out and the host is ending this
+            # session: answer as terminated rather than dispatch into a
+            # message loop that is going away.
+            if not self._terminated:
+                await self.terminate()
+            await self._handle_request(scope, receive, send)
+            return
+
+        # A request in flight (an open GET stream included) holds the session:
+        # the idle countdown is suspended while any is being served and
+        # restarts when the last one completes.
+        self._requests_in_flight += 1
+        self.idle_scope.deadline = math.inf
+        try:
+            await self._handle_request(scope, receive, send)
+        finally:
+            self._requests_in_flight -= 1
+            if not self._requests_in_flight:
+                self.idle_scope.deadline = anyio.current_time() + self._idle_timeout
+
+    async def _handle_request(self, scope: Scope, receive: Receive, send: Send) -> None:
         request = Request(scope, receive)
 
         # Validate request headers for DNS rebinding protection
@@ -643,8 +684,9 @@ async def _handle_post_request(self, scope: Scope, request: Request, receive: Re
                 except Exception:
                     logger.exception("SSE response error")
                     await sse_stream_writer.aclose()
-                    await sse_stream_reader.aclose()
                     await self._clean_up_memory_streams(request_id)
+                finally:
+                    await sse_stream_reader.aclose()
 
         except Exception as err:
             logger.exception("Error handling POST request")
@@ -747,9 +789,10 @@ async def standalone_sse_writer():
             await response(request.scope, request.receive, send)
         except Exception:
             logger.exception("Error in standalone SSE response")
+            await self._clean_up_memory_streams(GET_STREAM_KEY)
+        finally:
             await sse_stream_writer.aclose()
             await sse_stream_reader.aclose()
-            await self._clean_up_memory_streams(GET_STREAM_KEY)
 
     async def _handle_delete_request(self, request: Request, send: Send) -> None:  # pragma: no cover
         """Handle DELETE requests for explicit session termination."""
@@ -992,6 +1035,8 @@ async def connect(
         Yields:
             Tuple of (read_stream, write_stream) for bidirectional communication
         """
+        if self._idle_timeout is not None:
+            self.idle_scope = anyio.CancelScope()
 
         # Create the memory streams for this connection