Remove _effective_idle_timeout, use session_idle_timeout directly

The helper silently clamped the user's configured timeout to
retry_interval * 3, which is surprising. Users should set a timeout
that suits their deployment. Updated the docstring to note that the
timeout should comfortably exceed retry_interval when both are set.

Author: felixweinberger

src/mcp/server/streamable_http_manager.py

@@ -59,12 +59,11 @@ class StreamableHTTPSessionManager:
         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 / 1000 * 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.
+                            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__(
@@ -237,8 +236,8 @@ async def _handle_stateful_request(
             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:
-                transport.idle_scope.deadline = anyio.current_time() + self._effective_idle_timeout()
+            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
 
@@ -271,8 +270,7 @@ async def run_server(*, task_status: TaskStatus[None] = anyio.TASK_STATUS_IGNORE
                             # Incoming requests push the deadline forward.
                             idle_scope = anyio.CancelScope()
                             if self.session_idle_timeout is not None:
-                                timeout = self._effective_idle_timeout()
-                                idle_scope.deadline = anyio.current_time() + timeout
+                                idle_scope.deadline = anyio.current_time() + self.session_idle_timeout
                                 http_transport.idle_scope = idle_scope
 
                             with idle_scope:
@@ -332,19 +330,3 @@ 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 retry_interval is configured, clients periodically reconnect
-        to resume the event stream.  A gap of up to ``retry_interval`` between
-        connections is normal, not a sign of idleness.  We use a 3x multiplier
-        to tolerate up to two consecutive missed polls (network jitter, slow
-        client) before considering the session idle.
-        """
-        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

tests/issues/test_1283_idle_session_cleanup.py

@@ -155,25 +155,6 @@ async def test_terminate_idempotency():
         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)
-    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_sessions_persist():
     """When session_idle_timeout is None (default), sessions persist indefinitely."""