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,21 @@ 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).
+
+                            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 +81,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 +136,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 +150,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 +247,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 +267,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 +300,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 +327,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)

tests/server/test_streamable_http_manager.py

@@ -313,3 +313,173 @@ async def mock_receive():
         assert error_data["id"] == "server-error"
         assert error_data["error"]["code"] == INVALID_REQUEST
         assert error_data["error"]["message"] == "Session not found"
+
+
+# --- Idle timeout tests ---
+
+
+def _make_scope() -> dict:
+    return {
+        "type": "http",
+        "method": "POST",
+        "path": "/mcp",
+        "headers": [(b"content-type", b"application/json")],
+    }
+
+
+async def _idle_mock_receive() -> Message:  # pragma: no cover
+    return {"type": "http.request", "body": b"", "more_body": False}
+
+
+def _make_send(sent: list[Message]):
+    async def mock_send(message: Message) -> None:
+        sent.append(message)
+
+    return mock_send
+
+
+def _extract_session_id(sent_messages: list[Message]) -> str:
+    for msg in sent_messages:
+        if msg["type"] == "http.response.start":
+            for header_name, header_value in msg.get("headers", []):
+                if header_name.decode().lower() == MCP_SESSION_ID_HEADER.lower():
+                    return header_value.decode()
+    raise AssertionError("Session ID not found in response headers")
+
+
+def _make_blocking_run(stop_event: anyio.Event):
+    """Create a mock app.run that blocks until stop_event is set."""
+
+    async def blocking_run(*args, **kwargs):  # type: ignore[no-untyped-def]
+        await stop_event.wait()
+
+    return blocking_run
+
+
+@pytest.mark.anyio
+async def test_stateful_session_cleanup_on_idle_timeout():
+    """Session should be terminated and removed after being idle for the configured timeout."""
+    app = Server("test-idle-timeout")
+    stop = anyio.Event()
+    app.run = _make_blocking_run(stop)  # type: ignore[assignment]
+
+    manager = StreamableHTTPSessionManager(
+        app=app,
+        session_idle_timeout=0.15,
+    )
+
+    async with manager.run():
+        sent: list[Message] = []
+        await manager.handle_request(_make_scope(), _idle_mock_receive, _make_send(sent))
+        session_id = _extract_session_id(sent)
+
+        assert session_id in manager._server_instances
+        assert session_id in manager._last_activity
+
+        # Wait for reaper to fire
+        await anyio.sleep(0.4)
+
+        assert session_id not in manager._server_instances
+        assert session_id not in manager._last_activity
+
+        stop.set()
+
+
+@pytest.mark.anyio
+async def test_idle_timeout_reset_on_activity():
+    """Activity (requests) should reset the idle timer and prevent premature reaping."""
+    app = Server("test-idle-reset")
+    stop = anyio.Event()
+    app.run = _make_blocking_run(stop)  # type: ignore[assignment]
+
+    manager = StreamableHTTPSessionManager(
+        app=app,
+        session_idle_timeout=0.3,
+    )
+
+    async with manager.run():
+        sent: list[Message] = []
+        await manager.handle_request(_make_scope(), _idle_mock_receive, _make_send(sent))
+        session_id = _extract_session_id(sent)
+
+        # Simulate ongoing activity by bumping the timestamp
+        for _ in range(4):
+            await anyio.sleep(0.1)
+            manager._last_activity[session_id] = anyio.current_time()
+
+        # Session should still be alive
+        assert session_id in manager._server_instances
+
+        # Now stop activity and let timeout expire
+        await anyio.sleep(0.6)
+        assert session_id not in manager._server_instances
+
+        stop.set()
+
+
+@pytest.mark.anyio
+async def test_terminate_idempotency():
+    """Calling terminate() multiple times should be safe."""
+    transport = StreamableHTTPServerTransport(
+        mcp_session_id="test-idempotent",
+    )
+
+    # Set up minimal streams so terminate() has something to close
+    async with transport.connect():
+        await transport.terminate()
+        assert transport.is_terminated
+
+        # Second call should be a no-op (no exception)
+        await transport.terminate()
+        assert transport.is_terminated
+
+
+@pytest.mark.anyio
+async def test_idle_timeout_with_retry_interval():
+    """When retry_interval is set, effective timeout should account for polling gaps."""
+    app = Server("test-retry-interval")
+
+    # retry_interval = 5000ms = 5s → retry_seconds * 3 = 15s
+    # session_idle_timeout = 1s → effective = max(1, 15) = 15
+    manager = StreamableHTTPSessionManager(
+        app=app,
+        session_idle_timeout=1.0,
+        retry_interval=5000,
+    )
+    assert manager._effective_idle_timeout() == 15.0
+
+    # When retry_interval is small, session_idle_timeout should dominate
+    manager2 = StreamableHTTPSessionManager(
+        app=app,
+        session_idle_timeout=10.0,
+        retry_interval=100,  # 0.1s → 0.3s, less than 10
+    )
+    assert manager2._effective_idle_timeout() == 10.0
+
+    # No retry_interval → raw timeout
+    manager3 = StreamableHTTPSessionManager(
+        app=app,
+        session_idle_timeout=5.0,
+    )
+    assert manager3._effective_idle_timeout() == 5.0
+
+
+@pytest.mark.anyio
+async def test_no_idle_timeout_no_reaper():
+    """When session_idle_timeout is None, no reaper task should run and sessions persist."""
+    app = Server("test-no-timeout")
+    stop = anyio.Event()
+    app.run = _make_blocking_run(stop)  # type: ignore[assignment]
+
+    manager = StreamableHTTPSessionManager(app=app)
+
+    async with manager.run():
+        sent: list[Message] = []
+        await manager.handle_request(_make_scope(), _idle_mock_receive, _make_send(sent))
+        session_id = _extract_session_id(sent)
+
+        # Wait a while — session should never be reaped
+        await anyio.sleep(0.3)
+        assert session_id in manager._server_instances
+
+        stop.set()