feat: add idle timeout for StreamableHTTP sessions

Add session_idle_timeout parameter to StreamableHTTPSessionManager that
enables automatic cleanup of idle sessions, fixing the memory leak
described in issue #1283.

Key design decisions:
- Idle timeout logic lives in the session manager, not the transport.
  The manager already owns the session lifecycle, so this is the natural
  place for reaping. The transport stays unaware of timeout logic.
- A background reaper task scans _server_instances periodically and
  terminates sessions that have been idle longer than the threshold.
- When retry_interval (SSE polling) is configured, the effective timeout
  is at least retry_interval_seconds * 3 to avoid reaping sessions that
  are simply between polling reconnections.
- terminate() on the transport is now idempotent (early return if already
  terminated), making it safe to call from both the reaper and explicit
  DELETE requests.

Based on the approach from PR #1159 by @hopeful0, reworked to:
- Move idle tracking to the session manager instead of the transport
- Remove __aenter__/__aexit__ context manager and condition variables
- Account for retry_interval in the idle threshold
- Add comprehensive unit and integration tests

Github-Issue: #1283
Reported-by: hopeful0

Author: felixweinberger

src/mcp/server/streamable_http.py

@@ -773,8 +773,12 @@ async def terminate(self) -> None:
         """Terminate the current session, closing all streams.
 
         Once terminated, all requests with this session ID will receive 404 Not Found.
+        Calling this method multiple times is safe (idempotent).
         """
 
+        if self._terminated:
+            return
+
         self._terminated = True
         logger.info(f"Terminating session: {self.mcp_session_id}")
 

src/mcp/server/streamable_http_manager.py

@@ -38,6 +38,7 @@ class StreamableHTTPSessionManager:
     2. Resumability via an optional event store
     3. Connection management and lifecycle
     4. Request handling and transport setup
+    5. Idle session cleanup via optional timeout
 
     Important: Only one StreamableHTTPSessionManager instance should be created
     per application. The instance cannot be reused after its run() context has
@@ -55,6 +56,22 @@ class StreamableHTTPSessionManager:
         security_settings: Optional transport security settings.
         retry_interval: Retry interval in milliseconds to suggest to clients in SSE
                        retry field. Used for SSE polling behavior.
+        session_idle_timeout: Optional idle timeout in seconds for stateful sessions.
+                            If set, sessions that receive no HTTP requests for this
+                            duration will be automatically terminated and removed.
+                            When retry_interval is also set, the effective idle
+                            threshold is at least ``retry_interval_seconds * 3`` to
+                            avoid prematurely reaping sessions that are simply
+                            waiting for SSE polling reconnections. Default is None
+                            (no timeout). A value of 1800 (30 minutes) is
+                            recommended for most deployments.
+
+                            Note: The idle timer is based on incoming HTTP requests
+                            (POST, GET, DELETE), not on whether SSE connections are
+                            open. If clients maintain long-lived GET SSE streams
+                            without sending other requests, set this value higher
+                            than the longest expected SSE connection lifetime to
+                            avoid premature reaping.
     """
 
     def __init__(
@@ -65,17 +82,23 @@ def __init__(
         stateless: bool = False,
         security_settings: TransportSecuritySettings | None = None,
         retry_interval: int | None = None,
+        session_idle_timeout: float | None = None,
     ):
+        if session_idle_timeout is not None and session_idle_timeout <= 0:
+            raise ValueError("session_idle_timeout must be a positive number of seconds")
+
         self.app = app
         self.event_store = event_store
         self.json_response = json_response
         self.stateless = stateless
         self.security_settings = security_settings
         self.retry_interval = retry_interval
+        self.session_idle_timeout = session_idle_timeout
 
         # Session tracking (only used if not stateless)
         self._session_creation_lock = anyio.Lock()
         self._server_instances: dict[str, StreamableHTTPServerTransport] = {}
+        self._last_activity: dict[str, float] = {}
 
         # The task group will be set during lifespan
         self._task_group = None
@@ -114,6 +137,11 @@ async def lifespan(app: Starlette) -> AsyncIterator[None]:
             # Store the task group for later use
             self._task_group = tg
             logger.info("StreamableHTTP session manager started")
+
+            # Start idle session reaper if timeout is configured
+            if self.session_idle_timeout is not None:
+                tg.start_soon(self._idle_session_reaper)
+
             try:
                 yield  # Let the application run
             finally:
@@ -123,6 +151,7 @@ async def lifespan(app: Starlette) -> AsyncIterator[None]:
                 self._task_group = None
                 # Clear any remaining server instances
                 self._server_instances.clear()
+                self._last_activity.clear()
 
     async def handle_request(
         self,
@@ -219,6 +248,8 @@ async def _handle_stateful_request(
         if request_mcp_session_id is not None and request_mcp_session_id in self._server_instances:  # pragma: no cover
             transport = self._server_instances[request_mcp_session_id]
             logger.debug("Session already exists, handling request directly")
+            # Update activity timestamp for idle timeout tracking
+            self._last_activity[request_mcp_session_id] = anyio.current_time()
             await transport.handle_request(scope, receive, send)
             return
 
@@ -237,6 +268,7 @@ async def _handle_stateful_request(
 
                 assert http_transport.mcp_session_id is not None
                 self._server_instances[http_transport.mcp_session_id] = http_transport
+                self._last_activity[http_transport.mcp_session_id] = anyio.current_time()
                 logger.info(f"Created new transport with session ID: {new_session_id}")
 
                 # Define the server runner
@@ -269,6 +301,7 @@ async def run_server(*, task_status: TaskStatus[None] = anyio.TASK_STATUS_IGNORE
                                     "active instances."
                                 )
                                 del self._server_instances[http_transport.mcp_session_id]
+                                self._last_activity.pop(http_transport.mcp_session_id, None)
 
                 # Assert task group is not None for type checking
                 assert self._task_group is not None
@@ -295,3 +328,43 @@ async def run_server(*, task_status: TaskStatus[None] = anyio.TASK_STATUS_IGNORE
                 media_type="application/json",
             )
             await response(scope, receive, send)
+
+    def _effective_idle_timeout(self) -> float:
+        """Compute the effective idle timeout, accounting for retry_interval.
+
+        When SSE polling is configured via ``retry_interval`` (milliseconds),
+        the client may legitimately go quiet between polls.  The idle threshold
+        must be large enough so that normal polling gaps don't cause premature
+        session reaping.
+        """
+        assert self.session_idle_timeout is not None
+        timeout = self.session_idle_timeout
+        if self.retry_interval is not None:
+            retry_seconds = self.retry_interval / 1000.0
+            timeout = max(timeout, retry_seconds * 3)
+        return timeout
+
+    async def _idle_session_reaper(self) -> None:
+        """Background task that periodically terminates idle sessions."""
+        timeout = self._effective_idle_timeout()
+        scan_interval = min(timeout / 2, 30.0)
+        logger.info(f"Idle session reaper started (timeout={timeout}s, scan_interval={scan_interval}s)")
+
+        while True:
+            await anyio.sleep(scan_interval)
+            now = anyio.current_time()
+            # Snapshot keys to avoid mutation during iteration
+            for session_id in list(self._server_instances.keys()):
+                last = self._last_activity.get(session_id)
+                if last is None:
+                    continue  # pragma: no cover
+                if now - last > timeout:
+                    transport = self._server_instances.get(session_id)
+                    if transport is None:
+                        continue  # pragma: no cover
+                    logger.info(
+                        f"Terminating idle session {session_id} (idle for {now - last:.1f}s, timeout={timeout}s)"
+                    )
+                    await transport.terminate()
+                    self._server_instances.pop(session_id, None)
+                    self._last_activity.pop(session_id, None)