feat: add idle timeout for StreamableHTTP sessions (#1994)

Author: felixweinberger

CLAUDE.md

@@ -24,6 +24,11 @@ This document contains critical information about working with this codebase. Fo
    - Coverage: test edge cases and errors
    - New features require tests
    - Bug fixes require regression tests
+   - Avoid `anyio.sleep()` with a fixed duration to wait for async operations. Instead:
+     - Use `anyio.Event` — set it in the callback/handler, `await event.wait()` in the test
+     - For stream messages, use `await stream.receive()` instead of `sleep()` + `receive_nowait()`
+     - Exception: `sleep()` is appropriate when testing time-based features (e.g., timeouts)
+   - Wrap indefinite waits (`event.wait()`, `stream.receive()`) in `anyio.fail_after(5)` to prevent hangs
 
 - For commits fixing bugs or adding features based on user reports add:
 

src/mcp/server/streamable_http.py

@@ -180,6 +180,8 @@ def __init__(
         ] = {}
         self._sse_stream_writers: dict[RequestId, MemoryObjectSendStream[dict[str, str]]] = {}
         self._terminated = False
+        # Idle timeout cancel scope; managed by the session manager.
+        self.idle_scope: anyio.CancelScope | None = None
 
     @property
     def is_terminated(self) -> bool:
@@ -773,8 +775,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:  # pragma: no cover
+            return
+
         self._terminated = True
         logger.info(f"Terminating session: {self.mcp_session_id}")
 

src/mcp/server/streamable_http_manager.py

@@ -38,23 +38,28 @@ 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
     completed. If you need to restart the manager, create a new instance.
 
     Args:
         app: The MCP server instance
-        event_store: Optional event store for resumability support.
-                     If provided, enables resumable connections where clients
-                     can reconnect and receive missed events.
-                     If None, sessions are still tracked but not resumable.
+        event_store: Optional event store for resumability support. If provided, enables resumable connections
+            where clients can reconnect and receive missed events. If None, sessions are still tracked but not
+            resumable.
         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.
+        stateless: If True, creates a completely fresh transport for each request with no session tracking or
+            state persistence between requests.
         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.
+        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 configured, ensure the idle timeout comfortably exceeds the retry interval to
+            avoid reaping sessions during normal SSE polling gaps. Default is None (no timeout). A value of 1800
+            (30 minutes) is recommended for most deployments.
     """
 
     def __init__(
@@ -65,13 +70,20 @@ 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")
+        if stateless and session_idle_timeout is not None:
+            raise RuntimeError("session_idle_timeout is not supported in stateless mode")
+
         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()
@@ -219,6 +231,9 @@ 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")
+            # Push back idle deadline on activity
+            if transport.idle_scope is not None and self.session_idle_timeout is not None:
+                transport.idle_scope.deadline = anyio.current_time() + self.session_idle_timeout
             await transport.handle_request(scope, receive, send)
             return
 
@@ -245,19 +260,31 @@ async def run_server(*, task_status: TaskStatus[None] = anyio.TASK_STATUS_IGNORE
                         read_stream, write_stream = streams
                         task_status.started()
                         try:
-                            await self.app.run(
-                                read_stream,
-                                write_stream,
-                                self.app.create_initialization_options(),
-                                stateless=False,  # Stateful mode
-                            )
-                        except Exception as e:
-                            logger.error(
-                                f"Session {http_transport.mcp_session_id} crashed: {e}",
-                                exc_info=True,
-                            )
+                            # Use a cancel scope for idle timeout — when the
+                            # deadline passes the scope cancels app.run() and
+                            # execution continues after the ``with`` block.
+                            # Incoming requests push the deadline forward.
+                            idle_scope = anyio.CancelScope()
+                            if self.session_idle_timeout is not None:
+                                idle_scope.deadline = anyio.current_time() + self.session_idle_timeout
+                                http_transport.idle_scope = idle_scope
+
+                            with idle_scope:
+                                await self.app.run(
+                                    read_stream,
+                                    write_stream,
+                                    self.app.create_initialization_options(),
+                                    stateless=False,
+                                )
+
+                            if idle_scope.cancelled_caught:
+                                assert http_transport.mcp_session_id is not None
+                                logger.info(f"Session {http_transport.mcp_session_id} idle timeout")
+                                self._server_instances.pop(http_transport.mcp_session_id, None)
+                                await http_transport.terminate()
+                        except Exception:
+                            logger.exception(f"Session {http_transport.mcp_session_id} crashed")
                         finally:
-                            # Only remove from instances if not terminated
                             if (  # pragma: no branch
                                 http_transport.mcp_session_id
                                 and http_transport.mcp_session_id in self._server_instances

tests/server/test_streamable_http_manager.py

@@ -313,3 +313,80 @@ 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"
+
+
+@pytest.mark.anyio
+async def test_idle_session_is_reaped():
+    """After idle timeout fires, the session returns 404."""
+    app = Server("test-idle-reap")
+    manager = StreamableHTTPSessionManager(app=app, session_idle_timeout=0.05)
+
+    async with manager.run():
+        sent_messages: list[Message] = []
+
+        async def mock_send(message: Message):
+            sent_messages.append(message)
+
+        scope = {
+            "type": "http",
+            "method": "POST",
+            "path": "/mcp",
+            "headers": [(b"content-type", b"application/json")],
+        }
+
+        async def mock_receive():  # pragma: no cover
+            return {"type": "http.request", "body": b"", "more_body": False}
+
+        await manager.handle_request(scope, mock_receive, mock_send)
+
+        session_id = None
+        for msg in sent_messages:  # pragma: no branch
+            if msg["type"] == "http.response.start":  # pragma: no branch
+                for header_name, header_value in msg.get("headers", []):  # pragma: no branch
+                    if header_name.decode().lower() == MCP_SESSION_ID_HEADER.lower():
+                        session_id = header_value.decode()
+                        break
+                if session_id:  # pragma: no branch
+                    break
+
+        assert session_id is not None, "Session ID not found in response headers"
+
+        # Wait for the 50ms idle timeout to fire and cleanup to complete
+        await anyio.sleep(0.1)
+
+        # Verify via public API: old session ID now returns 404
+        response_messages: list[Message] = []
+
+        async def capture_send(message: Message):
+            response_messages.append(message)
+
+        scope_with_session = {
+            "type": "http",
+            "method": "POST",
+            "path": "/mcp",
+            "headers": [
+                (b"content-type", b"application/json"),
+                (b"mcp-session-id", session_id.encode()),
+            ],
+        }
+
+        await manager.handle_request(scope_with_session, mock_receive, capture_send)
+
+        response_start = next(
+            (msg for msg in response_messages if msg["type"] == "http.response.start"),
+            None,
+        )
+        assert response_start is not None
+        assert response_start["status"] == 404
+
+
+def test_session_idle_timeout_rejects_non_positive():
+    with pytest.raises(ValueError, match="positive number"):
+        StreamableHTTPSessionManager(app=Server("test"), session_idle_timeout=-1)
+    with pytest.raises(ValueError, match="positive number"):
+        StreamableHTTPSessionManager(app=Server("test"), session_idle_timeout=0)
+
+
+def test_session_idle_timeout_rejects_stateless():
+    with pytest.raises(RuntimeError, match="not supported in stateless"):
+        StreamableHTTPSessionManager(app=Server("test"), session_idle_timeout=30, stateless=True)