From 95db302656b02185307de92cf3d9935c12d2cb89 Mon Sep 17 00:00:00 2001
From: Max Isbey <224885523+maxisbey@users.noreply.github.com>
Date: Fri, 14 Aug 2026 12:05:20 +0000
Subject: [PATCH 1/5] Forget Streamable HTTP sessions when they end

The session manager kept a session's registry entry after the client
ended it with DELETE (the per-session task's cleanup skipped terminated
transports), and a request without a session ID that was refused
(anything but a valid initialize: wrong Accept, malformed JSON, a
non-initialize message, GET/DELETE) still left a registered transport
with a running server task behind it.

Now the manager drops the entry as soon as the transport is terminated,
the per-session task forgets the session and terminates its transport
however the loop ended, and a provisional session whose opening request
was answered with an error is discarded before the request returns. A
follow-up request on a deleted session is answered by the manager
("Session not found", 404) rather than by the dead transport.
---
 src/mcp/server/streamable_http_manager.py     |  67 +++--
 .../transports/test_hosting_session.py        |  14 +-
 tests/server/test_streamable_http_manager.py  | 244 ++++++++++++------
 tests/server/test_streamable_http_router.py   |  16 ++
 tests/shared/test_streamable_http.py          |  27 +-
 5 files changed, 253 insertions(+), 115 deletions(-)

diff --git a/src/mcp/server/streamable_http_manager.py b/src/mcp/server/streamable_http_manager.py
index e9a7d9629b..246083c204 100644
--- a/src/mcp/server/streamable_http_manager.py
+++ b/src/mcp/server/streamable_http_manager.py
@@ -14,7 +14,7 @@
 from mcp_types.version import HANDSHAKE_PROTOCOL_VERSIONS
 from starlette.requests import Request
 from starlette.responses import Response
-from starlette.types import Receive, Scope, Send
+from starlette.types import ASGIApp, Message, Receive, Scope, Send
 
 from mcp.server._streamable_http_modern import handle_modern_request
 from mcp.server.auth.middleware.bearer_auth import AuthenticatedUser, AuthorizationContext, authorization_context
@@ -278,6 +278,10 @@ async def _handle_stateful_request(self, scope: Scope, receive: Receive, send: S
             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  # pragma: no cover
             await transport.handle_request(scope, receive, send)
+            if transport.is_terminated:
+                # The client ended the session (DELETE): forget it now rather
+                # than when its server task winds down.
+                self._forget_session(request_mcp_session_id)
             return
 
         if request_mcp_session_id is None:
@@ -327,33 +331,37 @@ async def run_server(*, task_status: TaskStatus[None] = anyio.TASK_STATUS_IGNORE
                                 )
 
                             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)
-                                self._session_owners.pop(http_transport.mcp_session_id, None)
-                                await http_transport.terminate()
+                                logger.info(f"Session {new_session_id} idle timeout")
                         except Exception:
-                            logger.exception(f"Session {http_transport.mcp_session_id} crashed")
+                            logger.exception(f"Session {new_session_id} crashed")
                         finally:
-                            if (  # pragma: no branch
-                                http_transport.mcp_session_id
-                                and http_transport.mcp_session_id in self._server_instances
-                                and not http_transport.is_terminated
-                            ):
-                                logger.info(
-                                    "Cleaning up crashed session "
-                                    f"{http_transport.mcp_session_id} from active instances."
-                                )
-                                del self._server_instances[http_transport.mcp_session_id]
-                                self._session_owners.pop(http_transport.mcp_session_id, None)
+                            # However the session ended (client DELETE, idle
+                            # timeout, crash), stop tracking it and make sure the
+                            # transport refuses anything that still reaches it.
+                            self._forget_session(new_session_id)
+                            if not http_transport.is_terminated:
+                                await http_transport.terminate()
 
                 # Assert task group is not None for type checking
                 assert self._task_group is not None
                 # Start the server task
                 await self._task_group.start(run_server)
 
-                # Handle the HTTP request and return the response
-                await http_transport.handle_request(scope, receive, send)
+                # Handle the HTTP request and return the response. Without a
+                # session ID only an initialize request can succeed, so if this
+                # one was refused nothing was established: forget the session
+                # again rather than keep it (and its server task) around.
+                established = False
+                try:
+                    status = await _send_and_report_status(http_transport.handle_request, scope, receive, send)
+                    established = status is not None and status < 400
+                finally:
+                    if not established:  # pragma: no branch
+                        # Refused, failed or cancelled before a session was
+                        # established: nothing to keep.
+                        self._forget_session(new_session_id)
+                        with anyio.CancelScope(shield=True):
+                            await http_transport.terminate()
         else:
             # Unknown or expired session ID - return 404 per MCP spec
             # TODO(L62): Align error code once spec clarifies
@@ -367,6 +375,25 @@ async def run_server(*, task_status: TaskStatus[None] = anyio.TASK_STATUS_IGNORE
             )
             await response(scope, receive, send)
 
+    def _forget_session(self, session_id: str) -> None:
+        """Stop tracking a session; requests naming it are answered 404 from then on."""
+        self._server_instances.pop(session_id, None)
+        self._session_owners.pop(session_id, None)
+
+
+async def _send_and_report_status(app: ASGIApp, scope: Scope, receive: Receive, send: Send) -> int | None:
+    """Run `app` for one request and return the HTTP status it answered with (None if it sent no response)."""
+    status: int | None = None
+
+    async def watch_status(message: Message) -> None:
+        nonlocal status
+        if message["type"] == "http.response.start":
+            status = message["status"]
+        await send(message)
+
+    await app(scope, receive, watch_status)
+    return status
+
 
 class StreamableHTTPASGIApp:
     """ASGI application for Streamable HTTP server transport."""
diff --git a/tests/interaction/transports/test_hosting_session.py b/tests/interaction/transports/test_hosting_session.py
index 23c8da1580..6aa2a921fd 100644
--- a/tests/interaction/transports/test_hosting_session.py
+++ b/tests/interaction/transports/test_hosting_session.py
@@ -107,23 +107,15 @@ async def test_delete_terminates_the_session_and_subsequent_requests_return_404(
         delete = await http.delete("/mcp", headers=base_headers(session_id=session_id))
         assert delete.status_code == 200
 
-        # The manager keeps the terminated transport registered, so the next request reaches the
-        # transport's own _terminated check rather than the manager's unknown-session path.
-        assert session_id in manager._server_instances
+        # The manager forgets a terminated session, so from then on the ID is simply unknown.
+        assert session_id not in manager._server_instances
         post = await http.post(
             "/mcp",
             json={"jsonrpc": "2.0", "id": 2, "method": "tools/list"},
             headers=base_headers(session_id=session_id),
         )
         assert (post.status_code, post.json()) == snapshot(
-            (
-                404,
-                {
-                    "jsonrpc": "2.0",
-                    "id": None,
-                    "error": {"code": -32600, "message": "Not Found: Session has been terminated"},
-                },
-            )
+            (404, {"jsonrpc": "2.0", "id": None, "error": {"code": -32600, "message": "Session not found"}})
         )
 
 
diff --git a/tests/server/test_streamable_http_manager.py b/tests/server/test_streamable_http_manager.py
index 1c0f88a62f..9b0ae62300 100644
--- a/tests/server/test_streamable_http_manager.py
+++ b/tests/server/test_streamable_http_manager.py
@@ -10,7 +10,8 @@
 import httpx2
 import pytest
 from mcp_types import INVALID_REQUEST, ListToolsResult, PaginatedRequestParams
-from starlette.types import Message, Scope
+from mcp_types.version import LATEST_HANDSHAKE_VERSION
+from starlette.types import Message, Receive, Scope, Send
 
 from mcp import Client
 from mcp.client.streamable_http import streamable_http_client
@@ -20,6 +21,22 @@
 from mcp.server.streamable_http import MCP_SESSION_ID_HEADER, StreamableHTTPServerTransport
 from mcp.server.streamable_http_manager import DEFAULT_MAX_REQUEST_BODY_SIZE, StreamableHTTPSessionManager
 
+_JSON_HEADERS = {"accept": "application/json, text/event-stream", "content-type": "application/json"}
+
+_INITIALIZE_BODY = json.dumps(
+    {
+        "jsonrpc": "2.0",
+        "id": 1,
+        "method": "initialize",
+        "params": {
+            "protocolVersion": LATEST_HANDSHAKE_VERSION,
+            "capabilities": {},
+            "clientInfo": {"name": "test", "version": "0"},
+        },
+    }
+).encode()
+"""A wire-level initialize request: the only request that may open a session."""
+
 
 @pytest.mark.anyio
 async def test_run_can_only_be_called_once():
@@ -420,50 +437,30 @@ def emit(self, record: logging.LogRecord) -> None:
             self.reaped.set()
 
 
-@pytest.mark.anyio
-async def test_idle_session_is_reaped(caplog: pytest.LogCaptureFixture, request: pytest.FixtureRequest):
-    """After idle timeout fires, the session returns 404."""
-    app = Server("test-idle-reap")
-    manager = StreamableHTTPSessionManager(app=app, session_idle_timeout=0.05)
+def _observe_idle_timeout(caplog: pytest.LogCaptureFixture, request: pytest.FixtureRequest) -> _IdleTimeoutObserver:
+    """Install an observer for the manager's "idle timeout" log record for the rest of the test.
 
-    # The reap is observed through the manager's own "idle timeout" log record: the manager pops
-    # the session synchronously after emitting it, before its next await, so a waiter woken by
-    # the record always finds the session gone. caplog.set_level enables INFO so it is created.
+    The manager pops the session synchronously after emitting that record, before its next await,
+    so a waiter woken by it always finds the session gone. caplog.set_level enables INFO so the
+    record is created.
+    """
     observer = _IdleTimeoutObserver()
     manager_logger = logging.getLogger(streamable_http_manager.__name__)
     manager_logger.addHandler(observer)
     request.addfinalizer(lambda: manager_logger.removeHandler(observer))
     caplog.set_level(logging.INFO, logger=streamable_http_manager.__name__)
+    return observer
 
-    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():
-            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
+@pytest.mark.anyio
+async def test_idle_session_is_reaped(caplog: pytest.LogCaptureFixture, request: pytest.FixtureRequest):
+    """After idle timeout fires, the session returns 404."""
+    app = Server("test-idle-reap")
+    manager = StreamableHTTPSessionManager(app=app, session_idle_timeout=0.05)
+    observer = _observe_idle_timeout(caplog, request)
 
-        assert session_id is not None, "Session ID not found in response headers"
+    async with manager.run():
+        session_id = await _open_session(manager, None)
 
         # Wait for the 50ms idle timeout to fire and the session to be unregistered. Re-requesting
         # the session to poll for the 404 would push its idle deadline forward and keep it alive.
@@ -471,29 +468,7 @@ async def mock_receive():
             await observer.reaped.wait()
 
         # 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
+        assert await _request_session(manager, session_id, None) == 404
 
 
 def test_session_idle_timeout_rejects_non_positive():
@@ -508,6 +483,111 @@ def test_session_idle_timeout_rejects_stateless():
         StreamableHTTPSessionManager(app=Server("test"), session_idle_timeout=30, stateless=True)
 
 
+@pytest.mark.anyio
+async def test_deleted_session_is_forgotten() -> None:
+    """A client DELETE ends the session and the manager stops tracking it; the ID is unknown afterwards."""
+    manager = StreamableHTTPSessionManager(app=Server("test-delete"))
+    async with manager.run():
+        session_id = await _open_session(manager, None)
+        assert session_id in manager._server_instances
+
+        assert await _request_session(manager, session_id, None, method="DELETE") == 200
+        assert session_id not in manager._server_instances
+        response_start, response_body = await _call(manager, _request_scope(session_id=session_id))
+        assert response_start["status"] == 404
+        assert json.loads(response_body) == {
+            "jsonrpc": "2.0",
+            "id": None,
+            "error": {"code": INVALID_REQUEST, "message": "Session not found"},
+        }
+
+
+@pytest.mark.anyio
+async def test_opening_request_that_fails_leaves_no_session() -> None:
+    """If serving the request that would open a session raises, the provisional session is discarded
+    there and then rather than left registered with its server task running."""
+    manager = StreamableHTTPSessionManager(app=Server("test-failed-open"))
+    async with manager.run():
+        with (
+            patch.object(StreamableHTTPServerTransport, "handle_request", AsyncMock(side_effect=RuntimeError("boom"))),
+            pytest.raises(RuntimeError, match="boom"),
+            anyio.fail_after(5),
+        ):
+            await _call(manager, _request_scope(), _INITIALIZE_BODY)
+        assert manager._server_instances == {}
+        assert manager._session_owners == {}
+
+
+@pytest.mark.anyio
+async def test_opening_request_that_is_cancelled_leaves_no_session() -> None:
+    """If the request that would open a session is cancelled while it is being served (the client went
+    away), the provisional session is discarded rather than left registered."""
+    manager = StreamableHTTPSessionManager(app=Server("test-cancelled-open"))
+    entered = anyio.Event()
+
+    async def hang(self: StreamableHTTPServerTransport, scope: Scope, receive: Receive, send: Send) -> None:
+        entered.set()
+        await anyio.sleep_forever()
+
+    opening_request = anyio.CancelScope()
+
+    async def open_session() -> None:
+        with opening_request:
+            await _call(manager, _request_scope(), _INITIALIZE_BODY)
+
+    async with manager.run():
+        with patch.object(StreamableHTTPServerTransport, "handle_request", hang):
+            async with anyio.create_task_group() as tg:
+                tg.start_soon(open_session)
+                with anyio.fail_after(5):
+                    await entered.wait()
+                assert len(manager._server_instances) == 1
+                opening_request.cancel()
+            assert manager._server_instances == {}
+            assert manager._session_owners == {}
+
+
+@pytest.mark.anyio
+@pytest.mark.parametrize(
+    ("method", "headers", "body", "expected_status"),
+    [
+        ("POST", _JSON_HEADERS, b'{"jsonrpc": "2.0", "id": 1, "method": "tools/list"}', 400),
+        ("POST", _JSON_HEADERS, b'{"jsonrpc": "2.0", "method": "notifications/initialized"}', 400),
+        ("POST", _JSON_HEADERS, b"{not json", 400),
+        ("POST", _JSON_HEADERS | {"accept": "text/plain"}, _INITIALIZE_BODY, 406),
+        ("GET", {"accept": "text/event-stream"}, b"", 400),
+        ("DELETE", _JSON_HEADERS, b"", 400),
+        ("PATCH", _JSON_HEADERS, b"", 405),
+    ],
+    ids=[
+        "non-initialize-request",
+        "notification",
+        "malformed-json",
+        "unacceptable-accept-header",
+        "get-without-session",
+        "delete-without-session",
+        "unsupported-method",
+    ],
+)
+async def test_refused_opening_request_leaves_no_session(
+    method: str, headers: dict[str, str], body: bytes, expected_status: int
+) -> None:
+    """Only an accepted initialize opens a session: a request without a session ID that is answered with an
+    error leaves nothing registered once the manager has answered it."""
+    manager = StreamableHTTPSessionManager(app=Server("test-refused"))
+    scope: Scope = {
+        "type": "http",
+        "method": method,
+        "path": "/mcp",
+        "headers": [(name.encode(), value.encode()) for name, value in headers.items()],
+    }
+    async with manager.run():
+        response_start, _ = await _call(manager, scope, body)
+        assert response_start["status"] == expected_status
+        assert manager._server_instances == {}
+        assert manager._session_owners == {}
+
+
 def _user(client_id: str, subject: str | None = None, issuer: str | None = None) -> AuthenticatedUser:
     """Build the scope["user"] value that AuthenticationMiddleware would set for this principal."""
     claims = {"iss": issuer} if issuer is not None else None
@@ -535,19 +615,33 @@ def _request_scope(
     return scope
 
 
-async def _open_session(manager: StreamableHTTPSessionManager, user: AuthenticatedUser | None) -> str:
-    """Create a new session as `user` and return its session ID."""
+async def _call(manager: StreamableHTTPSessionManager, scope: Scope, body: bytes = b"") -> tuple[Message, bytes]:
+    """Drive one request through the manager in process; return its `http.response.start` message and body."""
     sent_messages: list[Message] = []
+    body_delivered = False
 
-    async def mock_send(message: Message) -> None:
+    async def send(message: Message) -> None:
         sent_messages.append(message)
 
-    async def mock_receive() -> Message:
-        return {"type": "http.request", "body": b"", "more_body": False}
+    async def receive() -> Message:
+        # Deliver the body once, then block like a client holding the connection
+        # open; a streaming response ends when the server closes it.
+        nonlocal body_delivered
+        if body_delivered:
+            await anyio.sleep_forever()
+        body_delivered = True
+        return {"type": "http.request", "body": body, "more_body": False}
+
+    await manager.handle_request(scope, receive, send)
+    response_start = next(msg for msg in sent_messages if msg["type"] == "http.response.start")
+    response_body = b"".join(msg.get("body", b"") for msg in sent_messages if msg["type"] == "http.response.body")
+    return response_start, response_body
 
-    await manager.handle_request(_request_scope(user=user), mock_receive, mock_send)
 
-    response_start = next(msg for msg in sent_messages if msg["type"] == "http.response.start")
+async def _open_session(manager: StreamableHTTPSessionManager, user: AuthenticatedUser | None) -> str:
+    """Create a new session as `user` with an initialize request and return its session ID."""
+    response_start, _ = await _call(manager, _request_scope(user=user), _INITIALIZE_BODY)
+    assert response_start["status"] == 200
     headers = dict(response_start.get("headers", []))
     return headers[MCP_SESSION_ID_HEADER.encode()].decode()
 
@@ -556,26 +650,14 @@ async def _request_session(
     manager: StreamableHTTPSessionManager, session_id: str, user: AuthenticatedUser | None, method: str = "POST"
 ) -> int:
     """Send a request for an existing session as `user` and return the response status."""
-    sent_messages: list[Message] = []
-
-    async def mock_send(message: Message) -> None:
-        sent_messages.append(message)
-
-    async def mock_receive() -> Message:
-        return {"type": "http.request", "body": b"", "more_body": False}
-
-    await manager.handle_request(
-        _request_scope(session_id=session_id, user=user, method=method), mock_receive, mock_send
-    )
-
-    response_start = next(msg for msg in sent_messages if msg["type"] == "http.response.start")
+    response_start, _ = await _call(manager, _request_scope(session_id=session_id, user=user, method=method))
     return response_start["status"]
 
 
 @pytest.fixture
 async def manager_with_live_session():
-    """A running manager around a real `Server`. Sessions remain registered until
-    `manager.run()` exits because `Server.run` blocks waiting for an initialize message."""
+    """A running manager around a real `Server`. Sessions are opened with a real initialize and stay
+    registered until `manager.run()` exits because nothing in these tests ends them."""
     manager = StreamableHTTPSessionManager(app=Server("test-session-credentials"))
     async with manager.run():
         yield manager
diff --git a/tests/server/test_streamable_http_router.py b/tests/server/test_streamable_http_router.py
index 07aa063499..0c5796c1f5 100644
--- a/tests/server/test_streamable_http_router.py
+++ b/tests/server/test_streamable_http_router.py
@@ -141,3 +141,19 @@ async def test_json_post_answers_500_when_session_terminates_mid_request() -> No
 
     assert post.sent[0]["type"] == "http.response.start"
     assert post.sent[0]["status"] == 500
+
+
+@pytest.mark.anyio
+async def test_terminated_transport_answers_404() -> None:
+    """A request that still reaches a transport after its session was terminated is answered 404."""
+    transport = StreamableHTTPServerTransport(mcp_session_id="sid")
+    post = _AsgiPost(
+        b'{"jsonrpc":"2.0","id":"req-1","method":"ping"}',
+        [(b"accept", b"application/json, text/event-stream"), (b"content-type", b"application/json")],
+    )
+    async with transport.connect():
+        await transport.terminate()
+        await transport.handle_request(post.scope, post.receive, post.send)
+
+    assert post.sent[0]["type"] == "http.response.start"
+    assert post.sent[0]["status"] == 404
diff --git a/tests/shared/test_streamable_http.py b/tests/shared/test_streamable_http.py
index aeef25a278..4252ca8b2d 100644
--- a/tests/shared/test_streamable_http.py
+++ b/tests/shared/test_streamable_http.py
@@ -7,6 +7,7 @@
 from __future__ import annotations as _annotations
 
 import json
+import logging
 import time
 from collections.abc import AsyncIterator
 from contextlib import asynccontextmanager
@@ -599,6 +600,26 @@ def test_streamable_http_transport_init_validation() -> None:
         StreamableHTTPServerTransport(mcp_session_id="test\n")
 
 
+@pytest.mark.anyio
+async def test_transport_reports_stream_closure_when_host_exits_without_terminating(
+    caplog: pytest.LogCaptureFixture,
+) -> None:
+    """A host that leaves connect() without terminating the transport closes the streams under the
+    message router, which reports it rather than passing it off as a client disconnect."""
+    caplog.set_level(logging.ERROR, logger="mcp.server.streamable_http")
+    transport = StreamableHTTPServerTransport(mcp_session_id="valid-id")
+
+    async with transport.connect():
+        pass
+
+    assert not transport.is_terminated
+    assert [
+        record.getMessage()
+        for record in caplog.records
+        if record.name == "mcp.server.streamable_http" and record.levelno == logging.ERROR
+    ] == ["Unexpected closure of read stream in message router"]
+
+
 @pytest.mark.anyio
 async def test_session_termination(basic_app: Starlette) -> None:
     """DELETE terminates the session, after which requests for it return 404."""
@@ -639,7 +660,7 @@ async def test_session_termination(basic_app: Starlette) -> None:
             json={"jsonrpc": "2.0", "method": "ping", "id": 2},
         )
         assert response.status_code == 404
-        assert "Session has been terminated" in response.text
+        assert response.json()["error"]["message"] == "Session not found"
 
 
 @pytest.mark.anyio
@@ -1048,7 +1069,7 @@ async def test_streamable_http_client_session_termination(basic_app: Starlette)
                 with pytest.raises(MCPError) as exc_info:  # pragma: no branch
                     await session.list_tools()
                 assert exc_info.value.error.code == INVALID_REQUEST
-                assert "terminated" in exc_info.value.error.message.lower()
+                assert exc_info.value.error.message == "Session not found"
 
 
 @pytest.mark.anyio
@@ -1111,7 +1132,7 @@ async def mock_delete(self: httpx2.AsyncClient, *args: Any, **kwargs: Any) -> ht
                 with pytest.raises(MCPError) as exc_info:  # pragma: no branch
                     await session.list_tools()
                 assert exc_info.value.error.code == INVALID_REQUEST
-                assert "terminated" in exc_info.value.error.message.lower()
+                assert exc_info.value.error.message == "Session not found"
 
 
 @pytest.mark.anyio

From 373e956b6feae5743290536e96471878d517f50f Mon Sep 17 00:00:00 2001
From: Max Isbey <224885523+maxisbey@users.noreply.github.com>
Date: Fri, 14 Aug 2026 12:05:48 +0000
Subject: [PATCH 2/5] Expire idle Streamable HTTP sessions by default

`session_idle_timeout` was opt-in (default None), so at stock settings a
stateful session that its client never deleted stayed registered, with
its server task and streams, until the process exited. The docstring
already recommended 1800 seconds; make that the default
(DEFAULT_SESSION_IDLE_TIMEOUT) so sessions nobody is using are
reclaimed after 30 minutes. `None` keeps the previous behaviour.

"Idle" is now measured from the moment the session's last in-flight
request completes rather than from the arrival of the last request:
the transport takes an `idle_timeout` and owns the countdown, holding
it while any request (an open GET stream included) is being served and
restarting it when the last one finishes. A connected client, or a call
that runs longer than the timeout, therefore never loses its session;
a client that goes quiet with no stream open gets 404 on its next
request and initializes again, as the spec describes.

The timeout is simply unused in stateless mode, which keeps no
sessions, so constructing a stateless manager with a timeout no longer
raises.
---
 src/mcp/server/streamable_http.py            |  40 ++++-
 src/mcp/server/streamable_http_manager.py    |  42 +++--
 tests/server/test_streamable_http_manager.py | 171 ++++++++++++++++++-
 tests/shared/test_streamable_http.py         |  27 +++
 4 files changed, 247 insertions(+), 33 deletions(-)

diff --git a/src/mcp/server/streamable_http.py b/src/mcp/server/streamable_http.py
index 1a4e9939a4..8eee255b8c 100644
--- a/src/mcp/server/streamable_http.py
+++ b/src/mcp/server/streamable_http.py
@@ -7,6 +7,7 @@
 """
 
 import logging
+import math
 import re
 from abc import ABC, abstractmethod
 from collections.abc import AsyncGenerator, Awaitable, Callable
@@ -167,6 +168,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.
 
@@ -187,12 +189,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 enters `idle_scope`
+                         (available once `connect()` has been entered) around the session's
+                         message loop to end 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 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 idle_timeout <= 0:
+            raise ValueError("idle_timeout must be a positive number of seconds")
 
         self.mcp_session_id = mcp_session_id
         self.is_json_response_enabled = is_json_response_enabled
@@ -208,8 +220,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:
@@ -458,6 +473,23 @@ async def _clean_up_memory_streams(self, request_id: RequestId) -> None:
 
     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
+
+        # 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
@@ -793,7 +825,7 @@ async def _handle_delete_request(self, request: Request, send: Send) -> None:
             await response(request.scope, request.receive, send)
             return
 
-        if not await self._validate_request_headers(request, send):  # pragma: no cover
+        if not await self._validate_request_headers(request, send):
             return
 
         await self.terminate()
@@ -995,6 +1027,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
 
diff --git a/src/mcp/server/streamable_http_manager.py b/src/mcp/server/streamable_http_manager.py
index 246083c204..1b2b11a38a 100644
--- a/src/mcp/server/streamable_http_manager.py
+++ b/src/mcp/server/streamable_http_manager.py
@@ -5,7 +5,7 @@
 import contextlib
 import logging
 from collections.abc import AsyncIterator
-from typing import TYPE_CHECKING, Any
+from typing import TYPE_CHECKING, Any, Final
 from uuid import uuid4
 
 import anyio
@@ -34,6 +34,9 @@
 
 logger = logging.getLogger(__name__)
 
+DEFAULT_SESSION_IDLE_TIMEOUT: Final = 30 * 60
+"""Default idle period in seconds after which a stateful Streamable HTTP session is closed (30 minutes)."""
+
 
 class StreamableHTTPSessionManager:
     """Manages StreamableHTTP sessions with optional resumability via event store.
@@ -45,7 +48,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
+    5. Idle session cleanup
 
     Important: Only one StreamableHTTPSessionManager instance should be created
     per application. The instance cannot be reused after its run() context has
@@ -62,11 +65,12 @@ 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 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.
+        session_idle_timeout: Idle timeout in seconds for stateful sessions. A session that has had no HTTP
+            request in flight for this long (no request being served, no open GET stream) is terminated and
+            removed; its ID then answers 404 and the client has to initialize a new session. When retry_interval
+            is also configured, ensure the idle timeout comfortably exceeds the retry interval to avoid reaping
+            sessions during normal SSE polling gaps. Defaults to 1800 (30 minutes); None disables the timeout so
+            sessions live until the client deletes them or the manager shuts down. Unused in stateless mode.
         max_request_body_size: Maximum size in bytes for Streamable HTTP request bodies. Requests that
             exceed this limit receive a 413 response before parsing or session creation. Defaults to 4 MiB.
     """
@@ -79,13 +83,11 @@ def __init__(
         stateless: bool = False,
         security_settings: TransportSecuritySettings | None = None,
         retry_interval: int | None = None,
-        session_idle_timeout: float | None = None,
+        session_idle_timeout: float | None = DEFAULT_SESSION_IDLE_TIMEOUT,
         max_request_body_size: int = DEFAULT_MAX_REQUEST_BODY_SIZE,
     ):
         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")
         if max_request_body_size <= 0:
             raise ValueError("max_request_body_size must be a positive number of bytes")
 
@@ -274,9 +276,6 @@ async def _handle_stateful_request(self, scope: Scope, receive: Receive, send: S
                 await response(scope, receive, send)
                 return
             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  # pragma: no cover
             await transport.handle_request(scope, receive, send)
             if transport.is_terminated:
                 # The client ended the session (DELETE): forget it now rather
@@ -295,6 +294,7 @@ async def _handle_stateful_request(self, scope: Scope, receive: Receive, send: S
                     event_store=self.event_store,  # May be None (no resumability)
                     security_settings=self.security_settings,
                     retry_interval=self.retry_interval,
+                    idle_timeout=self.session_idle_timeout,
                 )
 
                 assert http_transport.mcp_session_id is not None
@@ -309,15 +309,13 @@ async def run_server(*, task_status: TaskStatus[None] = anyio.TASK_STATUS_IGNORE
                         read_stream, write_stream = streams
                         task_status.started()
                         try:
-                            # Use a cancel scope for idle timeout — when the
-                            # deadline passes the scope cancels the loop 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
-
+                            # The transport cancels its idle scope once no request
+                            # has been in flight for `session_idle_timeout`; that
+                            # ends the loop and execution continues after the
+                            # `with` block. Without a timeout there is nothing to fire.
+                            idle_scope = http_transport.idle_scope
+                            if idle_scope is None:
+                                idle_scope = anyio.CancelScope()
                             with idle_scope:
                                 # Drive via `serve_loop` (not `Server.run()`) so the
                                 # manager's already-entered lifespan is reused
diff --git a/tests/server/test_streamable_http_manager.py b/tests/server/test_streamable_http_manager.py
index 9b0ae62300..701249e19e 100644
--- a/tests/server/test_streamable_http_manager.py
+++ b/tests/server/test_streamable_http_manager.py
@@ -2,15 +2,25 @@
 
 import json
 import logging
-from collections.abc import Iterator
+from collections.abc import AsyncIterator, Iterator
+from contextlib import asynccontextmanager
 from typing import Any
 from unittest.mock import AsyncMock, patch
 
 import anyio
 import httpx2
 import pytest
-from mcp_types import INVALID_REQUEST, ListToolsResult, PaginatedRequestParams
+from mcp_types import (
+    INVALID_REQUEST,
+    CallToolRequestParams,
+    CallToolResult,
+    ListToolsResult,
+    PaginatedRequestParams,
+    TextContent,
+)
 from mcp_types.version import LATEST_HANDSHAKE_VERSION
+from starlette.applications import Starlette
+from starlette.routing import Mount
 from starlette.types import Message, Receive, Scope, Send
 
 from mcp import Client
@@ -19,7 +29,15 @@
 from mcp.server.auth.middleware.bearer_auth import AuthenticatedUser
 from mcp.server.auth.provider import AccessToken
 from mcp.server.streamable_http import MCP_SESSION_ID_HEADER, StreamableHTTPServerTransport
-from mcp.server.streamable_http_manager import DEFAULT_MAX_REQUEST_BODY_SIZE, StreamableHTTPSessionManager
+from mcp.server.streamable_http_manager import (
+    DEFAULT_MAX_REQUEST_BODY_SIZE,
+    DEFAULT_SESSION_IDLE_TIMEOUT,
+    StreamableHTTPSessionManager,
+)
+from tests.interaction.transports import StreamingASGITransport
+
+# The in-process app is mounted at this origin purely so URLs are well-formed; nothing listens here.
+BASE_URL = "http://127.0.0.1:8000"
 
 _JSON_HEADERS = {"accept": "application/json, text/event-stream", "content-type": "application/json"}
 
@@ -452,6 +470,14 @@ def _observe_idle_timeout(caplog: pytest.LogCaptureFixture, request: pytest.Fixt
     return observer
 
 
+@asynccontextmanager
+async def _served(manager: StreamableHTTPSessionManager) -> AsyncIterator[httpx2.AsyncClient]:
+    """Run `manager` behind an in-process HTTP client whose responses stream as they are produced."""
+    app = Starlette(routes=[Mount("/", app=manager.handle_request)])
+    async with manager.run(), httpx2.AsyncClient(transport=StreamingASGITransport(app), base_url=BASE_URL) as http:
+        yield http
+
+
 @pytest.mark.anyio
 async def test_idle_session_is_reaped(caplog: pytest.LogCaptureFixture, request: pytest.FixtureRequest):
     """After idle timeout fires, the session returns 404."""
@@ -471,6 +497,129 @@ async def test_idle_session_is_reaped(caplog: pytest.LogCaptureFixture, request:
         assert await _request_session(manager, session_id, None) == 404
 
 
+@pytest.mark.anyio
+async def test_request_in_flight_holds_the_session_open(
+    caplog: pytest.LogCaptureFixture, request: pytest.FixtureRequest
+) -> None:
+    """A session does not expire while one of its requests is still being served, however long that takes;
+    the idle period is counted from the moment its last request completes."""
+    tool_started = anyio.Event()
+    release_tool = anyio.Event()
+
+    async def handle_call_tool(ctx: ServerRequestContext, params: CallToolRequestParams) -> CallToolResult:
+        tool_started.set()
+        await release_tool.wait()
+        return CallToolResult(content=[TextContent(type="text", text="done")])
+
+    app = Server("test-in-flight", on_call_tool=handle_call_tool)
+    manager = StreamableHTTPSessionManager(app=app, session_idle_timeout=0.05)
+    observer = _observe_idle_timeout(caplog, request)
+
+    async with _served(manager) as http:
+        initialize = await http.post("/mcp", content=_INITIALIZE_BODY, headers=_JSON_HEADERS)
+        assert initialize.status_code == 200
+        session_id = initialize.headers[MCP_SESSION_ID_HEADER]
+        call_tool_body: dict[str, Any] = {
+            "jsonrpc": "2.0",
+            "id": 2,
+            "method": "tools/call",
+            "params": {"name": "slow", "arguments": {}},
+        }
+        responses: list[httpx2.Response] = []
+
+        async def call_tool() -> None:
+            responses.append(
+                await http.post(
+                    "/mcp", json=call_tool_body, headers=_JSON_HEADERS | {MCP_SESSION_ID_HEADER: session_id}
+                )
+            )
+
+        async with anyio.create_task_group() as tg:
+            tg.start_soon(call_tool)
+            with anyio.fail_after(5):
+                await tool_started.wait()
+            # Several idle periods pass while the call is being served: the session stays.
+            await anyio.sleep(0.2)
+            assert session_id in manager._server_instances
+            assert not observer.reaped.is_set()
+            release_tool.set()
+
+        assert responses[0].status_code == 200
+        assert '"done"' in responses[0].text
+
+        # Nothing is in flight any more, so the idle period now runs out.
+        with anyio.fail_after(5):
+            await observer.reaped.wait()
+        assert session_id not in manager._server_instances
+
+
+@pytest.mark.anyio
+async def test_open_event_stream_holds_the_session_open(
+    caplog: pytest.LogCaptureFixture, request: pytest.FixtureRequest
+) -> None:
+    """A client listening on the session's GET stream keeps the session, even if it sends nothing;
+    once the stream closes the idle period runs out and the session is gone."""
+    manager = StreamableHTTPSessionManager(app=Server("test-get-stream"), session_idle_timeout=0.05)
+    observer = _observe_idle_timeout(caplog, request)
+
+    async with _served(manager) as http:
+        initialize = await http.post("/mcp", content=_INITIALIZE_BODY, headers=_JSON_HEADERS)
+        assert initialize.status_code == 200
+        session_id = initialize.headers[MCP_SESSION_ID_HEADER]
+
+        get_headers = {"accept": "text/event-stream", MCP_SESSION_ID_HEADER: session_id}
+        async with http.stream("GET", "/mcp", headers=get_headers) as stream:
+            assert stream.status_code == 200
+            await anyio.sleep(0.2)
+            assert session_id in manager._server_instances
+            assert not observer.reaped.is_set()
+
+        with anyio.fail_after(5):
+            await observer.reaped.wait()
+        assert session_id not in manager._server_instances
+        followup = await http.post("/mcp", json={"jsonrpc": "2.0", "id": 2, "method": "ping"}, headers=get_headers)
+        assert followup.status_code == 404
+
+
+@pytest.mark.anyio
+async def test_request_completing_under_an_open_event_stream_does_not_start_the_countdown(
+    caplog: pytest.LogCaptureFixture, request: pytest.FixtureRequest
+) -> None:
+    """A request that completes while the session's GET stream is still open does not start the idle
+    period: the stream is still in flight, so the countdown only begins once it closes too."""
+    manager = StreamableHTTPSessionManager(app=Server("test-get-stream-and-post"), session_idle_timeout=0.05)
+    observer = _observe_idle_timeout(caplog, request)
+
+    async with _served(manager) as http:
+        initialize = await http.post("/mcp", content=_INITIALIZE_BODY, headers=_JSON_HEADERS)
+        assert initialize.status_code == 200
+        session_id = initialize.headers[MCP_SESSION_ID_HEADER]
+
+        get_headers = {"accept": "text/event-stream", MCP_SESSION_ID_HEADER: session_id}
+        async with http.stream("GET", "/mcp", headers=get_headers) as stream:
+            assert stream.status_code == 200
+            ping = await http.post(
+                "/mcp",
+                json={"jsonrpc": "2.0", "id": 2, "method": "ping"},
+                headers=_JSON_HEADERS | {MCP_SESSION_ID_HEADER: session_id},
+            )
+            assert ping.status_code == 200
+            # Several idle periods pass after the ping completed; the open stream still holds the session.
+            await anyio.sleep(0.25)
+            assert session_id in manager._server_instances
+            assert not observer.reaped.is_set()
+
+        with anyio.fail_after(5):
+            await observer.reaped.wait()
+        assert session_id not in manager._server_instances
+
+
+def test_session_idle_timeout_defaults_to_thirty_minutes() -> None:
+    """Stateful sessions expire after 30 minutes without a request in flight unless configured otherwise."""
+    manager = StreamableHTTPSessionManager(app=Server("test"))
+    assert manager.session_idle_timeout == DEFAULT_SESSION_IDLE_TIMEOUT == 30 * 60
+
+
 def test_session_idle_timeout_rejects_non_positive():
     with pytest.raises(ValueError, match="positive number"):
         StreamableHTTPSessionManager(app=Server("test"), session_idle_timeout=-1)
@@ -478,15 +627,21 @@ def test_session_idle_timeout_rejects_non_positive():
         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)
+@pytest.mark.anyio
+async def test_session_idle_timeout_is_unused_in_stateless_mode() -> None:
+    """Stateless mode keeps no sessions, so the idle timeout is accepted and simply has nothing to expire."""
+    manager = StreamableHTTPSessionManager(app=Server("test"), session_idle_timeout=30, stateless=True)
+    async with manager.run():
+        response_start, _ = await _call(manager, _request_scope(), _INITIALIZE_BODY)
+        assert response_start["status"] == 200
+        assert manager._server_instances == {}
 
 
 @pytest.mark.anyio
-async def test_deleted_session_is_forgotten() -> None:
+@pytest.mark.parametrize("session_idle_timeout", [DEFAULT_SESSION_IDLE_TIMEOUT, None])
+async def test_deleted_session_is_forgotten(session_idle_timeout: float | None) -> None:
     """A client DELETE ends the session and the manager stops tracking it; the ID is unknown afterwards."""
-    manager = StreamableHTTPSessionManager(app=Server("test-delete"))
+    manager = StreamableHTTPSessionManager(app=Server("test-delete"), session_idle_timeout=session_idle_timeout)
     async with manager.run():
         session_id = await _open_session(manager, None)
         assert session_id in manager._server_instances
diff --git a/tests/shared/test_streamable_http.py b/tests/shared/test_streamable_http.py
index 4252ca8b2d..897f897c23 100644
--- a/tests/shared/test_streamable_http.py
+++ b/tests/shared/test_streamable_http.py
@@ -10,6 +10,7 @@
 import logging
 import time
 from collections.abc import AsyncIterator
+from concurrent.futures import ThreadPoolExecutor
 from contextlib import asynccontextmanager
 from dataclasses import dataclass, field
 from typing import Any
@@ -600,6 +601,32 @@ def test_streamable_http_transport_init_validation() -> None:
         StreamableHTTPServerTransport(mcp_session_id="test\n")
 
 
+@pytest.mark.parametrize("idle_timeout", [0, -1])
+def test_streamable_http_transport_rejects_non_positive_idle_timeout(idle_timeout: float) -> None:
+    """A transport's idle timeout must be a positive number of seconds; without one it never expires."""
+    with pytest.raises(ValueError, match="positive number of seconds"):
+        StreamableHTTPServerTransport(mcp_session_id="valid-id", idle_timeout=idle_timeout)
+    assert StreamableHTTPServerTransport(mcp_session_id="valid-id").idle_scope is None
+
+
+def test_streamable_http_transport_with_idle_timeout_can_be_created_outside_an_event_loop() -> None:
+    """The idle scope is only created once connect() is entered, so a transport with a timeout can be
+    constructed without a running event loop."""
+    # A bare thread has no async context; this one does, courtesy of the suite's shared runner.
+    with ThreadPoolExecutor(max_workers=1) as pool:
+        transport = pool.submit(StreamableHTTPServerTransport, mcp_session_id="valid-id", idle_timeout=5).result()
+    assert transport.idle_scope is None
+
+
+@pytest.mark.anyio
+async def test_streamable_http_transport_creates_its_idle_scope_on_connect() -> None:
+    """Entering connect() creates the idle scope the host enters around the session's message loop."""
+    transport = StreamableHTTPServerTransport(mcp_session_id="valid-id", idle_timeout=5)
+    async with transport.connect():
+        assert isinstance(transport.idle_scope, anyio.CancelScope)
+        await transport.terminate()
+
+
 @pytest.mark.anyio
 async def test_transport_reports_stream_closure_when_host_exits_without_terminating(
     caplog: pytest.LogCaptureFixture,

From ae2daca69bba7b27c0eafbfdd972498185bbf6cf Mon Sep 17 00:00:00 2001
From: Max Isbey <224885523+maxisbey@users.noreply.github.com>
Date: Fri, 14 Aug 2026 12:06:26 +0000
Subject: [PATCH 3/5] Cap concurrent Streamable HTTP sessions and expose the
 session limits on the server factories

Add `max_sessions` (DEFAULT_MAX_SESSIONS = 10_000, `None` for no
limit) to StreamableHTTPSessionManager: while that many stateful
sessions are open, a request that would open another is answered 503
with a JSON-RPC error body and nothing is allocated; existing sessions
are untouched and room frees up as they end or expire. This matches the
Ruby SDK's defaults (the C# SDK uses the same 10 000 figure).

`session_idle_timeout` and `max_sessions` are accepted by
`Server.streamable_http_app()`, `MCPServer.streamable_http_app()`,
`run_streamable_http_async()` and `run(transport="streamable-http")`,
the same way `max_request_body_size` is, so applications can tune or
disable them without reaching into `session_manager` after the fact.
Docs: run/index.md options list, run/legacy-clients.md session cost,
troubleshooting.md.
---
 docs/run/index.md                            |  6 +++
 docs/run/legacy-clients.md                   |  7 ++++
 docs/troubleshooting.md                      |  8 ++--
 src/mcp/server/lowlevel/server.py            | 11 ++++-
 src/mcp/server/mcpserver/server.py           | 16 +++++++-
 src/mcp/server/streamable_http_manager.py    | 42 ++++++++++++--------
 tests/docs_src/test_asgi.py                  | 16 +++++++-
 tests/docs_src/test_legacy_clients.py        | 16 ++++++++
 tests/server/test_streamable_http_manager.py | 37 +++++++++++++++++
 9 files changed, 136 insertions(+), 23 deletions(-)

diff --git a/docs/run/index.md b/docs/run/index.md
index da54dc31a5..48fc90c0b8 100644
--- a/docs/run/index.md
+++ b/docs/run/index.md
@@ -70,6 +70,12 @@ Each transport has its own keyword arguments, all on `run()`:
 * `max_request_body_size`: largest accepted request body in bytes. Defaults to 4 MiB; larger requests
   receive HTTP 413 before parsing or session creation. Raise it only when legitimate MCP messages
   exceed that size.
+* `session_idle_timeout`: how long, in seconds, a [legacy](legacy-clients.md) (session-based)
+  client's session may sit with no request in flight before the server closes it. Defaults to 1800
+  (30 minutes); `None` keeps sessions until the client deletes them. A client with an open `GET`
+  stream or a request still being answered is never idle.
+* `max_sessions`: how many such sessions one app holds at once. Defaults to 10 000; while that many
+  are open, a request that would open another gets HTTP 503. `None` removes the limit.
 * `event_store`, `retry_interval`, `transport_security`: resumability and DNS-rebinding protection. They can wait, until you deploy somewhere other than localhost; **[Deploy & scale](deploy.md)** covers `transport_security`.
 
 !!! warning
diff --git a/docs/run/legacy-clients.md b/docs/run/legacy-clients.md
index a1c0f76007..04398cab09 100644
--- a/docs/run/legacy-clients.md
+++ b/docs/run/legacy-clients.md
@@ -56,6 +56,13 @@ On one worker that is invisible. On two, it is the whole problem: a request that
     events to a client reconnecting to the *same* session), not a session store. It never makes a
     session reachable from another process.
 
+The record is not kept forever. A client that ends its session (`DELETE`) frees it at once;
+a session that has had no request in flight for `session_idle_timeout` seconds (default 1800; an
+open `GET` stream or a request being answered counts as in flight) is closed, and its next request
+gets the same `404` a stray ID gets, so the client has to `initialize` again. Each worker process
+holds at most `max_sessions` of them (default 10 000) and answers `503` to a request that would
+open one more. Both are `run()` / `streamable_http_app()` options.
+
 ## The one knob: `stateless_http`
 
 If stickiness is a cost you refuse to pay, there is exactly one thing you can change.
diff --git a/docs/troubleshooting.md b/docs/troubleshooting.md
index b0e365caaa..664a9e67b7 100644
--- a/docs/troubleshooting.md
+++ b/docs/troubleshooting.md
@@ -246,7 +246,7 @@ app = Starlette(routes=[Mount("/", app=mcp.streamable_http_app())], lifespan=lif
 
 ## `MCPError: Session not found`
 
-The server does not recognise the `Mcp-Session-Id` your client sent, almost always because the server **restarted** (or you were routed to a different instance). Sessions live in that one process's memory.
+The server does not recognise the `Mcp-Session-Id` your client sent, because the server **restarted** (or you were routed to a different instance), or because the session **expired**: a legacy session with no request in flight for `session_idle_timeout` (30 minutes by default; an open `GET` stream or a request being answered counts as in flight) is closed, as is one the client ended with `DELETE`. Sessions live in that one process's memory.
 
 There is no server bug to find. The HTTP response is a `404` whose body *is* JSON-RPC, so, unlike the `421` above, the python `Client` shows you this one verbatim:
 
@@ -256,9 +256,9 @@ There is no server bug to find. The HTTP response is a `404` whose body *is* JSO
 
 The fix is to reconnect: leave the `async with Client(...)` block and enter a new one, which negotiates a fresh session. For a long-lived client, that means catching `MCPError` around your calls and reconnecting on this message rather than retrying inside a dead session.
 
-If it happens *without* a restart, you are running more than one worker without sticky sessions: each worker holds its own session table, so a request routed to the wrong one lands here. **[Deploy & scale](run/deploy.md)** and **[Serving legacy clients](run/legacy-clients.md)** own that story and its two fixes (sticky routing, or `stateless_http=True`).
+If it happens *without* a restart and without the client having gone quiet that long, you are running more than one worker without sticky sessions: each worker holds its own session table, so a request routed to the wrong one lands here. **[Deploy & scale](run/deploy.md)** and **[Serving legacy clients](run/legacy-clients.md)** own that story and its two fixes (sticky routing, or `stateless_http=True`).
 
-For the server operator, the matching log line is `Rejected request with unknown or expired session ID: <id>`. It is logged at `INFO`, so it is invisible at the usual `WARNING` threshold. Seeing it in bursts right after a deploy is normal; every connected client is reconnecting.
+For the server operator, the matching log line is `Rejected request with unknown or expired session ID: <id>`. It is logged at `INFO`, so it is invisible at the usual `WARNING` threshold. Seeing it in bursts right after a deploy is normal; every connected client is reconnecting. When the session expired instead, that line is preceded by `Session <id> idle timeout`, also at `INFO`.
 
 ## `MCPError: Method not found`
 
@@ -411,7 +411,7 @@ mcp = MCPServer("Weather", request_state_security=RequestStateSecurity(keys=[key
 * `Tool already exists:` in the server log is the only sign that two same-named tools collapsed into one.
 * One 421, three spellings: `Server returned an error response` (the python `Client`), `421 Misdirected Request` / `Invalid Host header` (everything else), `Invalid Host header: <host>` (the server log). Fix: `transport_security=TransportSecuritySettings(allowed_hosts=[...])`.
 * `Task group is not initialized` -> a mounted app whose host lifespan never entered `mcp.session_manager.run()`.
-* `Session not found` -> the server restarted; reconnect.
+* `Session not found` -> the server restarted or the session expired (`session_idle_timeout`); reconnect.
 * `Cannot send 'elicitation/create': ... no back-channel ...` -> `ctx.elicit()` needs a server-to-client channel: a `2026-07-28` connection never has one, `stateless_http=True` takes away the legacy one, and `json_response=True` takes away the request-scoped one. Use a resolver (a legacy client also needs a server that keeps the channel). Its neighbour `Method not found` is a request for a method the other side's protocol revision doesn't have.
 * `Client did not declare the form elicitation capability ...` and `Elicitation not supported` -> the client is missing `elicitation_callback=`.
 * `Invalid or expired requestState` never says why on the wire. The server log does; `unknown key` means share `RequestStateSecurity(keys=[...])` across workers.
diff --git a/src/mcp/server/lowlevel/server.py b/src/mcp/server/lowlevel/server.py
index 4c327f4ece..6df5341e41 100644
--- a/src/mcp/server/lowlevel/server.py
+++ b/src/mcp/server/lowlevel/server.py
@@ -65,7 +65,12 @@ async def main():
 from mcp.server.models import InitializationOptions
 from mcp.server.runner import serve_dual_era_loop
 from mcp.server.streamable_http import EventStore
-from mcp.server.streamable_http_manager import StreamableHTTPASGIApp, StreamableHTTPSessionManager
+from mcp.server.streamable_http_manager import (
+    DEFAULT_MAX_SESSIONS,
+    DEFAULT_SESSION_IDLE_TIMEOUT,
+    StreamableHTTPASGIApp,
+    StreamableHTTPSessionManager,
+)
 from mcp.server.transport_security import DEFAULT_MAX_REQUEST_BODY_SIZE, TransportSecuritySettings
 from mcp.shared._stream_protocols import ReadStream, WriteStream
 from mcp.shared.exceptions import MCPDeprecationWarning
@@ -722,6 +727,8 @@ def streamable_http_app(
         event_store: EventStore | None = None,
         retry_interval: int | None = None,
         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,
         transport_security: TransportSecuritySettings | None = None,
         host: str = "127.0.0.1",
         auth: AuthSettings | None = None,
@@ -747,6 +754,8 @@ def streamable_http_app(
             stateless=stateless_http,
             security_settings=transport_security,
             max_request_body_size=max_request_body_size,
+            session_idle_timeout=session_idle_timeout,
+            max_sessions=max_sessions,
         )
         self._session_manager = session_manager
 
diff --git a/src/mcp/server/mcpserver/server.py b/src/mcp/server/mcpserver/server.py
index 93bef1655a..1f3e863dbd 100644
--- a/src/mcp/server/mcpserver/server.py
+++ b/src/mcp/server/mcpserver/server.py
@@ -93,7 +93,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.subscriptions import InMemorySubscriptionBus, ListenHandler, SubscriptionBus
 from mcp.server.transport_security import DEFAULT_MAX_REQUEST_BODY_SIZE, TransportSecuritySettings
 from mcp.shared.exceptions import MCPError
@@ -388,6 +392,8 @@ def run(
         event_store: EventStore | None = ...,
         retry_interval: int | None = ...,
         max_request_body_size: int = ...,
+        session_idle_timeout: float | None = ...,
+        max_sessions: int | None = ...,
         transport_security: TransportSecuritySettings | None = ...,
     ) -> None: ...
 
@@ -1106,6 +1112,8 @@ async def run_streamable_http_async(  # pragma: no cover
         event_store: EventStore | None = None,
         retry_interval: int | None = None,
         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,
         transport_security: TransportSecuritySettings | None = None,
     ) -> None:
         """Run the server using StreamableHTTP transport."""
@@ -1118,6 +1126,8 @@ async def run_streamable_http_async(  # pragma: no cover
             event_store=event_store,
             retry_interval=retry_interval,
             max_request_body_size=max_request_body_size,
+            session_idle_timeout=session_idle_timeout,
+            max_sessions=max_sessions,
             transport_security=transport_security,
             host=host,
         )
@@ -1270,6 +1280,8 @@ def streamable_http_app(
         event_store: EventStore | None = None,
         retry_interval: int | None = None,
         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,
         transport_security: TransportSecuritySettings | None = None,
         host: str = "127.0.0.1",
     ) -> Starlette:
@@ -1281,6 +1293,8 @@ def streamable_http_app(
             event_store=event_store,
             retry_interval=retry_interval,
             max_request_body_size=max_request_body_size,
+            session_idle_timeout=session_idle_timeout,
+            max_sessions=max_sessions,
             transport_security=transport_security,
             host=host,
             auth=self.settings.auth,
diff --git a/src/mcp/server/streamable_http_manager.py b/src/mcp/server/streamable_http_manager.py
index 1b2b11a38a..30222f6722 100644
--- a/src/mcp/server/streamable_http_manager.py
+++ b/src/mcp/server/streamable_http_manager.py
@@ -37,6 +37,9 @@
 DEFAULT_SESSION_IDLE_TIMEOUT: Final = 30 * 60
 """Default idle period in seconds after which a stateful Streamable HTTP session is closed (30 minutes)."""
 
+DEFAULT_MAX_SESSIONS: Final = 10_000
+"""Default maximum number of concurrent stateful Streamable HTTP sessions per session manager."""
+
 
 class StreamableHTTPSessionManager:
     """Manages StreamableHTTP sessions with optional resumability via event store.
@@ -73,6 +76,10 @@ class StreamableHTTPSessionManager:
             sessions live until the client deletes them or the manager shuts down. Unused in stateless mode.
         max_request_body_size: Maximum size in bytes for Streamable HTTP request bodies. Requests that
             exceed this limit receive a 413 response before parsing or session creation. Defaults to 4 MiB.
+        max_sessions: Maximum number of concurrent stateful sessions. While that many sessions are open, a
+            request that would open another one receives a 503 response; existing sessions are unaffected and
+            room frees up as they end or expire. Defaults to 10 000; None removes the limit. Unused in stateless
+            mode.
     """
 
     def __init__(
@@ -85,11 +92,14 @@ def __init__(
         retry_interval: int | None = None,
         session_idle_timeout: float | None = DEFAULT_SESSION_IDLE_TIMEOUT,
         max_request_body_size: int = DEFAULT_MAX_REQUEST_BODY_SIZE,
+        max_sessions: int | None = DEFAULT_MAX_SESSIONS,
     ):
         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 max_request_body_size <= 0:
             raise ValueError("max_request_body_size must be a positive number of bytes")
+        if max_sessions is not None and max_sessions <= 0:
+            raise ValueError("max_sessions must be a positive number of sessions or None")
 
         self.app = app
         self.event_store = event_store
@@ -99,6 +109,7 @@ def __init__(
         self.retry_interval = retry_interval
         self.session_idle_timeout = session_idle_timeout
         self.max_request_body_size = max_request_body_size
+        self.max_sessions = max_sessions
         self.asgi_app = RequestBodyLimitMiddleware(self._handle_request, max_request_body_size)
 
         # Session tracking (only used if not stateless)
@@ -265,15 +276,7 @@ async def _handle_stateful_request(self, scope: Scope, receive: Receive, send: S
                     "Rejecting request for session %s: credential does not match the one that created the session",
                     request_mcp_session_id[:64],
                 )
-                body = JSONRPCError(
-                    jsonrpc="2.0", id=None, error=ErrorData(code=INVALID_REQUEST, message="Session not found")
-                )
-                response = Response(
-                    body.model_dump_json(by_alias=True, exclude_unset=True),
-                    status_code=404,
-                    media_type="application/json",
-                )
-                await response(scope, receive, send)
+                await _error_response("Session not found", 404)(scope, receive, send)
                 return
             logger.debug("Session already exists, handling request directly")
             await transport.handle_request(scope, receive, send)
@@ -287,6 +290,11 @@ async def _handle_stateful_request(self, scope: Scope, receive: Receive, send: S
             # New session case
             logger.debug("Creating new transport")
             async with self._session_creation_lock:
+                if self.max_sessions is not None and len(self._server_instances) >= self.max_sessions:
+                    logger.warning("Refusing to open a new session: %d sessions are already open", self.max_sessions)
+                    await _error_response("Too many open sessions", 503)(scope, receive, send)
+                    return
+
                 new_session_id = uuid4().hex
                 http_transport = StreamableHTTPServerTransport(
                     mcp_session_id=new_session_id,
@@ -365,13 +373,7 @@ async def run_server(*, task_status: TaskStatus[None] = anyio.TASK_STATUS_IGNORE
             # TODO(L62): Align error code once spec clarifies
             # See: https://github.com/modelcontextprotocol/python-sdk/issues/1821
             logger.info(f"Rejected request with unknown or expired session ID: {request_mcp_session_id[:64]}")
-            body = JSONRPCError(
-                jsonrpc="2.0", id=None, error=ErrorData(code=INVALID_REQUEST, message="Session not found")
-            )
-            response = Response(
-                body.model_dump_json(by_alias=True, exclude_unset=True), status_code=404, media_type="application/json"
-            )
-            await response(scope, receive, send)
+            await _error_response("Session not found", 404)(scope, receive, send)
 
     def _forget_session(self, session_id: str) -> None:
         """Stop tracking a session; requests naming it are answered 404 from then on."""
@@ -379,6 +381,14 @@ def _forget_session(self, session_id: str) -> None:
         self._session_owners.pop(session_id, None)
 
 
+def _error_response(message: str, status_code: int) -> Response:
+    """A JSON-RPC error body (no request id) with the given HTTP status."""
+    body = JSONRPCError(jsonrpc="2.0", id=None, error=ErrorData(code=INVALID_REQUEST, message=message))
+    return Response(
+        body.model_dump_json(by_alias=True, exclude_unset=True), status_code=status_code, media_type="application/json"
+    )
+
+
 async def _send_and_report_status(app: ASGIApp, scope: Scope, receive: Receive, send: Send) -> int | None:
     """Run `app` for one request and return the HTTP status it answered with (None if it sent no response)."""
     status: int | None = None
diff --git a/tests/docs_src/test_asgi.py b/tests/docs_src/test_asgi.py
index ef237bc212..d241dc71f9 100644
--- a/tests/docs_src/test_asgi.py
+++ b/tests/docs_src/test_asgi.py
@@ -13,7 +13,7 @@
 
 from docs_src.asgi import tutorial001, tutorial002, tutorial003, tutorial004, tutorial005, tutorial006
 from mcp import Client
-from mcp.server import MCPServer
+from mcp.server import MCPServer, Server
 
 # See test_index.py for why this is a per-module mark and not a conftest hook.
 pytestmark = [pytest.mark.anyio, pytest.mark.filterwarnings("error::mcp.MCPDeprecationWarning")]
@@ -44,6 +44,8 @@ async def test_streamable_http_app_takes_runs_options_except_port() -> None:
         "event_store",
         "retry_interval",
         "max_request_body_size",
+        "session_idle_timeout",
+        "max_sessions",
         "transport_security",
         "host",
     }
@@ -68,6 +70,18 @@ async def test_streamable_http_app_applies_the_configured_request_body_limit() -
     assert response.status_code == 413
 
 
+async def test_streamable_http_app_applies_the_configured_session_limits() -> None:
+    """The documented `session_idle_timeout` and `max_sessions` options reach the session manager, from
+    both the high-level and the low-level factory."""
+    server = MCPServer("Notes")
+    server.streamable_http_app(session_idle_timeout=5, max_sessions=7)
+    assert (server.session_manager.session_idle_timeout, server.session_manager.max_sessions) == (5, 7)
+
+    lowlevel = Server("Notes")
+    lowlevel.streamable_http_app(session_idle_timeout=None, max_sessions=None)
+    assert (lowlevel.session_manager.session_idle_timeout, lowlevel.session_manager.max_sessions) == (None, None)
+
+
 async def test_mounting_at_the_root_keeps_the_default_path() -> None:
     """tutorial002: `Mount("/")` plus the default `streamable_http_path` leaves the endpoint at `/mcp`."""
     (mount,) = tutorial002.app.routes
diff --git a/tests/docs_src/test_legacy_clients.py b/tests/docs_src/test_legacy_clients.py
index 3018e4fcaa..f6f2f9061c 100644
--- a/tests/docs_src/test_legacy_clients.py
+++ b/tests/docs_src/test_legacy_clients.py
@@ -53,6 +53,8 @@ def test_streamable_http_app_has_no_era_knob() -> None:
         "event_store",
         "retry_interval",
         "max_request_body_size",
+        "session_idle_timeout",
+        "max_sessions",
         "transport_security",
         "host",
     }
@@ -76,6 +78,20 @@ async def test_a_legacy_session_is_minted_in_process_and_a_stray_session_id_is_a
         assert stray.status_code == 404
 
 
+def test_legacy_sessions_expire_and_are_capped_by_default() -> None:
+    """The cost section: a session record is dropped after 30 idle minutes and each worker process holds at most
+    10 000 of them, unless `run()` / `streamable_http_app()` say otherwise."""
+    server = MCPServer("Bookshop")
+    server.streamable_http_app()
+    assert server.session_manager.session_idle_timeout == 30 * 60
+    assert server.session_manager.max_sessions == 10_000
+
+    server = MCPServer("Bookshop")
+    server.streamable_http_app(session_idle_timeout=None, max_sessions=None)
+    assert server.session_manager.session_idle_timeout is None
+    assert server.session_manager.max_sessions is None
+
+
 async def test_stateless_http_never_mints_a_session() -> None:
     """The `stateless_http=True` section: the same legacy `initialize` no longer gets an `Mcp-Session-Id`."""
     app = MCPServer("Bookshop").streamable_http_app(stateless_http=True)
diff --git a/tests/server/test_streamable_http_manager.py b/tests/server/test_streamable_http_manager.py
index 701249e19e..ad915bbe29 100644
--- a/tests/server/test_streamable_http_manager.py
+++ b/tests/server/test_streamable_http_manager.py
@@ -31,6 +31,7 @@
 from mcp.server.streamable_http import MCP_SESSION_ID_HEADER, StreamableHTTPServerTransport
 from mcp.server.streamable_http_manager import (
     DEFAULT_MAX_REQUEST_BODY_SIZE,
+    DEFAULT_MAX_SESSIONS,
     DEFAULT_SESSION_IDLE_TIMEOUT,
     StreamableHTTPSessionManager,
 )
@@ -743,6 +744,42 @@ async def test_refused_opening_request_leaves_no_session(
         assert manager._session_owners == {}
 
 
+@pytest.mark.anyio
+async def test_new_session_is_refused_at_max_sessions() -> None:
+    """At the session limit a further initialize is answered 503 and opens nothing; room frees up as
+    sessions end."""
+    manager = StreamableHTTPSessionManager(app=Server("test-cap"), max_sessions=1)
+    async with manager.run():
+        first = await _open_session(manager, None)
+
+        response_start, response_body = await _call(manager, _request_scope(), _INITIALIZE_BODY)
+        assert response_start["status"] == 503
+        assert json.loads(response_body) == {
+            "jsonrpc": "2.0",
+            "id": None,
+            "error": {"code": INVALID_REQUEST, "message": "Too many open sessions"},
+        }
+        assert list(manager._server_instances) == [first]
+
+        assert await _request_session(manager, first, None, method="DELETE") == 200
+        second = await _open_session(manager, None)
+        assert list(manager._server_instances) == [second]
+
+
+def test_max_sessions_defaults_to_ten_thousand() -> None:
+    """A manager holds at most 10 000 concurrent stateful sessions unless configured otherwise."""
+    manager = StreamableHTTPSessionManager(app=Server("test"))
+    assert manager.max_sessions == DEFAULT_MAX_SESSIONS == 10_000
+    assert StreamableHTTPSessionManager(app=Server("test"), max_sessions=None).max_sessions is None
+
+
+@pytest.mark.parametrize("max_sessions", [0, -1])
+def test_max_sessions_rejects_non_positive_values(max_sessions: int) -> None:
+    with pytest.raises(ValueError) as exc_info:
+        StreamableHTTPSessionManager(app=Server("test"), max_sessions=max_sessions)
+    assert str(exc_info.value) == "max_sessions must be a positive number of sessions or None"
+
+
 def _user(client_id: str, subject: str | None = None, issuer: str | None = None) -> AuthenticatedUser:
     """Build the scope["user"] value that AuthenticationMiddleware would set for this principal."""
     claims = {"iss": issuer} if issuer is not None else None

From 1b50a125fe43bbb19d1c46936ae0359b314f4db8 Mon Sep 17 00:00:00 2001
From: Max Isbey <224885523+maxisbey@users.noreply.github.com>
Date: Wed, 26 Aug 2026 15:10:29 +0000
Subject: [PATCH 4/5] End Streamable HTTP sessions through one path

Review follow-ups. The session manager now discards a session --
forgets it and terminates its transport, shielded from cancellation --
through one helper, used when the client deletes the session, when the
session task ends, and when the request that opens the session is
refused, fails or is cancelled; that bracket now starts when the session
is registered, so a session whose
task cannot be started is discarded as well. A stateless transport is
terminated even when its request is cancelled, so the per-request task
always ends. A transport whose idle period has run out answers as
terminated instead of dispatching into a message loop that is shutting
down. Idle timeouts must be finite as well as positive.

Tests assert that discarded sessions' transports are terminated, cover
the two new cleanup paths and the expired-transport answer, and the
idle-hold tests check the suspended countdown directly and arm the
short timeout only once the holding request is in flight instead of
racing a timer.
---
 src/mcp/server/streamable_http.py            |  15 +-
 src/mcp/server/streamable_http_manager.py    |  79 ++++---
 tests/server/test_streamable_http_manager.py | 219 +++++++++++++------
 tests/shared/test_streamable_http.py         |  53 ++++-
 4 files changed, 253 insertions(+), 113 deletions(-)

diff --git a/src/mcp/server/streamable_http.py b/src/mcp/server/streamable_http.py
index 8eee255b8c..416dd9e2b4 100644
--- a/src/mcp/server/streamable_http.py
+++ b/src/mcp/server/streamable_http.py
@@ -199,12 +199,12 @@ def __init__(
 
         Raises:
             ValueError: If the session ID contains invalid characters, or if `idle_timeout`
-                        is not a positive number.
+                        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 idle_timeout <= 0:
-            raise ValueError("idle_timeout must be a positive number of seconds")
+        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
@@ -477,6 +477,15 @@ async def handle_request(self, scope: Scope, receive: Receive, send: Send) -> No
             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.
diff --git a/src/mcp/server/streamable_http_manager.py b/src/mcp/server/streamable_http_manager.py
index 30222f6722..e84be74bb7 100644
--- a/src/mcp/server/streamable_http_manager.py
+++ b/src/mcp/server/streamable_http_manager.py
@@ -4,6 +4,7 @@
 
 import contextlib
 import logging
+import math
 from collections.abc import AsyncIterator
 from typing import TYPE_CHECKING, Any, Final
 from uuid import uuid4
@@ -94,8 +95,8 @@ def __init__(
         max_request_body_size: int = DEFAULT_MAX_REQUEST_BODY_SIZE,
         max_sessions: int | None = DEFAULT_MAX_SESSIONS,
     ):
-        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 session_idle_timeout is not None and not (math.isfinite(session_idle_timeout) and session_idle_timeout > 0):
+            raise ValueError("session_idle_timeout must be a positive, finite number of seconds")
         if max_request_body_size <= 0:
             raise ValueError("max_request_body_size must be a positive number of bytes")
         if max_sessions is not None and max_sessions <= 0:
@@ -247,16 +248,15 @@ async def run_stateless_server(*, task_status: TaskStatus[None] = anyio.TASK_STA
                 except Exception:  # pragma: lax no cover
                     logger.exception("Stateless session crashed")
 
-        # Assert task group is not None for type checking
+        # The per-request server task only ends once the transport is
+        # terminated, so terminate it even if the request was cancelled.
         assert self._task_group is not None
-        # Start the server task
-        await self._task_group.start(run_stateless_server)
-
-        # Handle the HTTP request and return the response
-        await http_transport.handle_request(scope, receive, send)
-
-        # Terminate the transport after the request is handled
-        await http_transport.terminate()
+        try:
+            await self._task_group.start(run_stateless_server)
+            await http_transport.handle_request(scope, receive, send)
+        finally:
+            with anyio.CancelScope(shield=True):
+                await http_transport.terminate()
 
     async def _handle_stateful_request(self, scope: Scope, receive: Receive, send: Send) -> None:
         """Process request in stateful mode - maintaining session state between requests."""
@@ -283,7 +283,7 @@ async def _handle_stateful_request(self, scope: Scope, receive: Receive, send: S
             if transport.is_terminated:
                 # The client ended the session (DELETE): forget it now rather
                 # than when its server task winds down.
-                self._forget_session(request_mcp_session_id)
+                await self._discard_session(request_mcp_session_id, transport)
             return
 
         if request_mcp_session_id is None:
@@ -305,13 +305,6 @@ async def _handle_stateful_request(self, scope: Scope, receive: Receive, send: S
                     idle_timeout=self.session_idle_timeout,
                 )
 
-                assert http_transport.mcp_session_id is not None
-                if requestor is not None:
-                    self._session_owners[http_transport.mcp_session_id] = requestor
-                self._server_instances[http_transport.mcp_session_id] = http_transport
-                logger.info(f"Created new transport with session ID: {new_session_id}")
-
-                # Define the server runner
                 async def run_server(*, task_status: TaskStatus[None] = anyio.TASK_STATUS_IGNORED) -> None:
                     async with http_transport.connect() as streams:
                         read_stream, write_stream = streams
@@ -342,32 +335,28 @@ async def run_server(*, task_status: TaskStatus[None] = anyio.TASK_STATUS_IGNORE
                             logger.exception(f"Session {new_session_id} crashed")
                         finally:
                             # However the session ended (client DELETE, idle
-                            # timeout, crash), stop tracking it and make sure the
-                            # transport refuses anything that still reaches it.
-                            self._forget_session(new_session_id)
-                            if not http_transport.is_terminated:
-                                await http_transport.terminate()
-
-                # Assert task group is not None for type checking
-                assert self._task_group is not None
-                # Start the server task
-                await self._task_group.start(run_server)
-
-                # Handle the HTTP request and return the response. Without a
-                # session ID only an initialize request can succeed, so if this
-                # one was refused nothing was established: forget the session
-                # again rather than keep it (and its server task) around.
+                            # timeout, crash), discard it.
+                            await self._discard_session(new_session_id, http_transport)
+
+                if requestor is not None:
+                    self._session_owners[new_session_id] = requestor
+                self._server_instances[new_session_id] = http_transport
+                logger.info(f"Created new transport with session ID: {new_session_id}")
+
+                # Without a session ID only an initialize request can succeed,
+                # so if this one is refused, fails or is cancelled (or the
+                # session's server task cannot even be started) nothing was
+                # established: discard the session again rather than keep it
+                # (and its server task) around.
                 established = False
                 try:
+                    assert self._task_group is not None
+                    await self._task_group.start(run_server)
                     status = await _send_and_report_status(http_transport.handle_request, scope, receive, send)
                     established = status is not None and status < 400
                 finally:
                     if not established:  # pragma: no branch
-                        # Refused, failed or cancelled before a session was
-                        # established: nothing to keep.
-                        self._forget_session(new_session_id)
-                        with anyio.CancelScope(shield=True):
-                            await http_transport.terminate()
+                        await self._discard_session(new_session_id, http_transport)
         else:
             # Unknown or expired session ID - return 404 per MCP spec
             # TODO(L62): Align error code once spec clarifies
@@ -375,10 +364,18 @@ async def run_server(*, task_status: TaskStatus[None] = anyio.TASK_STATUS_IGNORE
             logger.info(f"Rejected request with unknown or expired session ID: {request_mcp_session_id[:64]}")
             await _error_response("Session not found", 404)(scope, receive, send)
 
-    def _forget_session(self, session_id: str) -> None:
-        """Stop tracking a session; requests naming it are answered 404 from then on."""
+    async def _discard_session(self, session_id: str, transport: StreamableHTTPServerTransport) -> None:
+        """Stop tracking the session and make sure its transport refuses anything that still reaches it.
+
+        The session is forgotten first, before any await, so its ID answers 404
+        from the moment this is called; terminating the transport is shielded so
+        it completes even while the caller is being cancelled.
+        """
         self._server_instances.pop(session_id, None)
         self._session_owners.pop(session_id, None)
+        if not transport.is_terminated:
+            with anyio.CancelScope(shield=True):
+                await transport.terminate()
 
 
 def _error_response(message: str, status_code: int) -> Response:
diff --git a/tests/server/test_streamable_http_manager.py b/tests/server/test_streamable_http_manager.py
index ad915bbe29..d281ebd62e 100644
--- a/tests/server/test_streamable_http_manager.py
+++ b/tests/server/test_streamable_http_manager.py
@@ -2,8 +2,9 @@
 
 import json
 import logging
+import math
 from collections.abc import AsyncIterator, Iterator
-from contextlib import asynccontextmanager
+from contextlib import asynccontextmanager, contextmanager
 from typing import Any
 from unittest.mock import AsyncMock, patch
 
@@ -21,7 +22,7 @@
 from mcp_types.version import LATEST_HANDSHAKE_VERSION
 from starlette.applications import Starlette
 from starlette.routing import Mount
-from starlette.types import Message, Receive, Scope, Send
+from starlette.types import ASGIApp, Message, Receive, Scope, Send
 
 from mcp import Client
 from mcp.client.streamable_http import streamable_http_client
@@ -322,19 +323,7 @@ async def test_stateless_requests_memory_cleanup():
     app = Server("test-stateless-real-cleanup")
     manager = StreamableHTTPSessionManager(app=app, stateless=True)
 
-    # Track created transport instances
-    created_transports: list[StreamableHTTPServerTransport] = []
-
-    # Patch StreamableHTTPServerTransport constructor to track instances
-
-    original_constructor = StreamableHTTPServerTransport
-
-    def track_transport(*args: Any, **kwargs: Any) -> StreamableHTTPServerTransport:
-        transport = original_constructor(*args, **kwargs)
-        created_transports.append(transport)
-        return transport
-
-    with patch.object(streamable_http_manager, "StreamableHTTPServerTransport", side_effect=track_transport):
+    with _created_transports() as created_transports:
         async with manager.run():
             # Send a simple request
             sent_messages: list[Message] = []
@@ -471,10 +460,29 @@ def _observe_idle_timeout(caplog: pytest.LogCaptureFixture, request: pytest.Fixt
     return observer
 
 
+@contextmanager
+def _created_transports() -> Iterator[list[StreamableHTTPServerTransport]]:
+    """Collect every transport a session manager creates while the context is open."""
+    created: list[StreamableHTTPServerTransport] = []
+
+    def create(*args: Any, **kwargs: Any) -> StreamableHTTPServerTransport:
+        transport = StreamableHTTPServerTransport(*args, **kwargs)
+        created.append(transport)
+        return transport
+
+    with patch.object(streamable_http_manager, "StreamableHTTPServerTransport", side_effect=create):
+        yield created
+
+
 @asynccontextmanager
-async def _served(manager: StreamableHTTPSessionManager) -> AsyncIterator[httpx2.AsyncClient]:
-    """Run `manager` behind an in-process HTTP client whose responses stream as they are produced."""
-    app = Starlette(routes=[Mount("/", app=manager.handle_request)])
+async def _served(
+    manager: StreamableHTTPSessionManager, endpoint: ASGIApp | None = None
+) -> AsyncIterator[httpx2.AsyncClient]:
+    """Run `manager` behind an in-process HTTP client whose responses stream as they are produced.
+
+    `endpoint` is the ASGI app mounted for it; by default the manager itself.
+    """
+    app = Starlette(routes=[Mount("/", app=endpoint or manager.handle_request)])
     async with manager.run(), httpx2.AsyncClient(transport=StreamingASGITransport(app), base_url=BASE_URL) as http:
         yield http
 
@@ -513,13 +521,15 @@ async def handle_call_tool(ctx: ServerRequestContext, params: CallToolRequestPar
         return CallToolResult(content=[TextContent(type="text", text="done")])
 
     app = Server("test-in-flight", on_call_tool=handle_call_tool)
-    manager = StreamableHTTPSessionManager(app=app, session_idle_timeout=0.05)
+    manager = StreamableHTTPSessionManager(app=app, session_idle_timeout=30)
     observer = _observe_idle_timeout(caplog, request)
 
     async with _served(manager) as http:
         initialize = await http.post("/mcp", content=_INITIALIZE_BODY, headers=_JSON_HEADERS)
         assert initialize.status_code == 200
         session_id = initialize.headers[MCP_SESSION_ID_HEADER]
+        transport = manager._server_instances[session_id]
+        session_headers = _JSON_HEADERS | {MCP_SESSION_ID_HEADER: session_id}
         call_tool_body: dict[str, Any] = {
             "jsonrpc": "2.0",
             "id": 2,
@@ -529,20 +539,16 @@ async def handle_call_tool(ctx: ServerRequestContext, params: CallToolRequestPar
         responses: list[httpx2.Response] = []
 
         async def call_tool() -> None:
-            responses.append(
-                await http.post(
-                    "/mcp", json=call_tool_body, headers=_JSON_HEADERS | {MCP_SESSION_ID_HEADER: session_id}
-                )
-            )
+            responses.append(await http.post("/mcp", json=call_tool_body, headers=session_headers))
 
         async with anyio.create_task_group() as tg:
             tg.start_soon(call_tool)
             with anyio.fail_after(5):
                 await tool_started.wait()
-            # Several idle periods pass while the call is being served: the session stays.
-            await anyio.sleep(0.2)
-            assert session_id in manager._server_instances
-            assert not observer.reaped.is_set()
+            # While the call is being served the idle countdown is suspended.
+            assert transport.idle_scope is not None and transport.idle_scope.deadline == math.inf
+            # From here on a short idle period, counted from the moment the call completes.
+            transport._idle_timeout = 0.05
             release_tool.set()
 
         assert responses[0].status_code == 200
@@ -552,6 +558,9 @@ async def call_tool() -> None:
         with anyio.fail_after(5):
             await observer.reaped.wait()
         assert session_id not in manager._server_instances
+        assert transport.is_terminated
+        followup = await http.post("/mcp", json={"jsonrpc": "2.0", "id": 3, "method": "ping"}, headers=session_headers)
+        assert followup.status_code == 404
 
 
 @pytest.mark.anyio
@@ -560,25 +569,29 @@ async def test_open_event_stream_holds_the_session_open(
 ) -> None:
     """A client listening on the session's GET stream keeps the session, even if it sends nothing;
     once the stream closes the idle period runs out and the session is gone."""
-    manager = StreamableHTTPSessionManager(app=Server("test-get-stream"), session_idle_timeout=0.05)
+    manager = StreamableHTTPSessionManager(app=Server("test-get-stream"), session_idle_timeout=30)
     observer = _observe_idle_timeout(caplog, request)
 
     async with _served(manager) as http:
         initialize = await http.post("/mcp", content=_INITIALIZE_BODY, headers=_JSON_HEADERS)
         assert initialize.status_code == 200
         session_id = initialize.headers[MCP_SESSION_ID_HEADER]
+        session_headers = _JSON_HEADERS | {MCP_SESSION_ID_HEADER: session_id}
 
         get_headers = {"accept": "text/event-stream", MCP_SESSION_ID_HEADER: session_id}
         async with http.stream("GET", "/mcp", headers=get_headers) as stream:
             assert stream.status_code == 200
-            await anyio.sleep(0.2)
-            assert session_id in manager._server_instances
-            assert not observer.reaped.is_set()
+            # The stream has been answered, so it is in flight: the idle countdown is suspended.
+            transport = manager._server_instances[session_id]
+            assert transport.idle_scope is not None and transport.idle_scope.deadline == math.inf
+            # From here on a short idle period, counted from the moment the stream closes.
+            transport._idle_timeout = 0.05
 
         with anyio.fail_after(5):
             await observer.reaped.wait()
         assert session_id not in manager._server_instances
-        followup = await http.post("/mcp", json={"jsonrpc": "2.0", "id": 2, "method": "ping"}, headers=get_headers)
+        assert transport.is_terminated
+        followup = await http.post("/mcp", json={"jsonrpc": "2.0", "id": 2, "method": "ping"}, headers=session_headers)
         assert followup.status_code == 404
 
 
@@ -588,31 +601,43 @@ async def test_request_completing_under_an_open_event_stream_does_not_start_the_
 ) -> None:
     """A request that completes while the session's GET stream is still open does not start the idle
     period: the stream is still in flight, so the countdown only begins once it closes too."""
-    manager = StreamableHTTPSessionManager(app=Server("test-get-stream-and-post"), session_idle_timeout=0.05)
+    manager = StreamableHTTPSessionManager(app=Server("test-get-stream-and-post"), session_idle_timeout=30)
     observer = _observe_idle_timeout(caplog, request)
+    session_post_served = anyio.Event()
 
-    async with _served(manager) as http:
+    async def endpoint(scope: Scope, receive: Receive, send: Send) -> None:
+        # Report once a POST for the open session has been served to the end,
+        # in-flight bookkeeping included.
+        await manager.handle_request(scope, receive, send)
+        if scope["method"] == "POST" and MCP_SESSION_ID_HEADER.encode() in dict(scope["headers"]):
+            session_post_served.set()
+
+    async with _served(manager, endpoint) as http:
         initialize = await http.post("/mcp", content=_INITIALIZE_BODY, headers=_JSON_HEADERS)
         assert initialize.status_code == 200
         session_id = initialize.headers[MCP_SESSION_ID_HEADER]
+        session_headers = _JSON_HEADERS | {MCP_SESSION_ID_HEADER: session_id}
 
         get_headers = {"accept": "text/event-stream", MCP_SESSION_ID_HEADER: session_id}
         async with http.stream("GET", "/mcp", headers=get_headers) as stream:
             assert stream.status_code == 200
-            ping = await http.post(
-                "/mcp",
-                json={"jsonrpc": "2.0", "id": 2, "method": "ping"},
-                headers=_JSON_HEADERS | {MCP_SESSION_ID_HEADER: session_id},
-            )
+            transport = manager._server_instances[session_id]
+            assert transport.idle_scope is not None and transport.idle_scope.deadline == math.inf
+            ping = await http.post("/mcp", json={"jsonrpc": "2.0", "id": 2, "method": "ping"}, headers=session_headers)
             assert ping.status_code == 200
-            # Several idle periods pass after the ping completed; the open stream still holds the session.
-            await anyio.sleep(0.25)
-            assert session_id in manager._server_instances
-            assert not observer.reaped.is_set()
+            with anyio.fail_after(5):
+                await session_post_served.wait()
+            # The ping has completed, but the open stream still suspends the idle countdown.
+            assert transport.idle_scope.deadline == math.inf
+            # From here on a short idle period, counted from the moment the stream closes.
+            transport._idle_timeout = 0.05
 
         with anyio.fail_after(5):
             await observer.reaped.wait()
         assert session_id not in manager._server_instances
+        assert transport.is_terminated
+        followup = await http.post("/mcp", json={"jsonrpc": "2.0", "id": 3, "method": "ping"}, headers=session_headers)
+        assert followup.status_code == 404
 
 
 def test_session_idle_timeout_defaults_to_thirty_minutes() -> None:
@@ -621,11 +646,12 @@ def test_session_idle_timeout_defaults_to_thirty_minutes() -> None:
     assert manager.session_idle_timeout == DEFAULT_SESSION_IDLE_TIMEOUT == 30 * 60
 
 
-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)
+@pytest.mark.parametrize("session_idle_timeout", [0, -1, float("inf"), float("nan")])
+def test_session_idle_timeout_rejects_invalid_values(session_idle_timeout: float) -> None:
+    """The idle timeout is a positive, finite number of seconds, or None for sessions that never expire."""
+    with pytest.raises(ValueError) as exc_info:
+        StreamableHTTPSessionManager(app=Server("test"), session_idle_timeout=session_idle_timeout)
+    assert str(exc_info.value) == "session_idle_timeout must be a positive, finite number of seconds"
 
 
 @pytest.mark.anyio
@@ -663,15 +689,20 @@ async def test_opening_request_that_fails_leaves_no_session() -> None:
     """If serving the request that would open a session raises, the provisional session is discarded
     there and then rather than left registered with its server task running."""
     manager = StreamableHTTPSessionManager(app=Server("test-failed-open"))
-    async with manager.run():
-        with (
-            patch.object(StreamableHTTPServerTransport, "handle_request", AsyncMock(side_effect=RuntimeError("boom"))),
-            pytest.raises(RuntimeError, match="boom"),
-            anyio.fail_after(5),
-        ):
-            await _call(manager, _request_scope(), _INITIALIZE_BODY)
-        assert manager._server_instances == {}
-        assert manager._session_owners == {}
+    with _created_transports() as transports:
+        async with manager.run():
+            with (
+                patch.object(
+                    StreamableHTTPServerTransport, "handle_request", AsyncMock(side_effect=RuntimeError("boom"))
+                ),
+                pytest.raises(RuntimeError, match="boom"),
+                anyio.fail_after(5),
+            ):
+                await _call(manager, _request_scope(), _INITIALIZE_BODY)
+            assert manager._server_instances == {}
+            assert manager._session_owners == {}
+            (transport,) = transports
+            assert transport.is_terminated
 
 
 @pytest.mark.anyio
@@ -691,8 +722,8 @@ async def open_session() -> None:
         with opening_request:
             await _call(manager, _request_scope(), _INITIALIZE_BODY)
 
-    async with manager.run():
-        with patch.object(StreamableHTTPServerTransport, "handle_request", hang):
+    with _created_transports() as transports, patch.object(StreamableHTTPServerTransport, "handle_request", hang):
+        async with manager.run():
             async with anyio.create_task_group() as tg:
                 tg.start_soon(open_session)
                 with anyio.fail_after(5):
@@ -701,6 +732,61 @@ async def open_session() -> None:
                 opening_request.cancel()
             assert manager._server_instances == {}
             assert manager._session_owners == {}
+            (transport,) = transports
+            assert transport.is_terminated
+
+
+@pytest.mark.anyio
+async def test_opening_request_whose_session_task_cannot_start_leaves_no_session() -> None:
+    """If the server task for a would-be session cannot be started, the provisional session is discarded
+    (forgotten, its transport terminated) rather than left registered without anything serving it."""
+    manager = StreamableHTTPSessionManager(app=Server("test-unstartable-open"))
+
+    @asynccontextmanager
+    async def connect_that_fails(self: StreamableHTTPServerTransport) -> AsyncIterator[None]:
+        raise RuntimeError("boom")
+        yield
+
+    with _created_transports() as transports:
+        async with manager.run():
+            with (
+                patch.object(StreamableHTTPServerTransport, "connect", connect_that_fails),
+                pytest.raises(RuntimeError, match="boom"),
+                anyio.fail_after(5),
+            ):
+                await _call(manager, _request_scope(), _INITIALIZE_BODY)
+            assert manager._server_instances == {}
+            assert manager._session_owners == {}
+            (transport,) = transports
+            assert transport.is_terminated
+
+
+@pytest.mark.anyio
+async def test_stateless_request_that_is_cancelled_still_terminates_its_transport() -> None:
+    """If a stateless request is cancelled while it is being served (the client went away), its transport
+    is terminated all the same, which is what ends the per-request server task."""
+    manager = StreamableHTTPSessionManager(app=Server("test-stateless-cancelled"), stateless=True)
+    entered = anyio.Event()
+
+    async def hang(self: StreamableHTTPServerTransport, scope: Scope, receive: Receive, send: Send) -> None:
+        entered.set()
+        await anyio.sleep_forever()
+
+    stateless_request = anyio.CancelScope()
+
+    async def make_request() -> None:
+        with stateless_request:
+            await _call(manager, _request_scope(), _INITIALIZE_BODY)
+
+    with _created_transports() as transports, patch.object(StreamableHTTPServerTransport, "handle_request", hang):
+        async with manager.run():
+            async with anyio.create_task_group() as tg:
+                tg.start_soon(make_request)
+                with anyio.fail_after(5):
+                    await entered.wait()
+                stateless_request.cancel()
+            (transport,) = transports
+            assert transport.is_terminated
 
 
 @pytest.mark.anyio
@@ -737,11 +823,14 @@ async def test_refused_opening_request_leaves_no_session(
         "path": "/mcp",
         "headers": [(name.encode(), value.encode()) for name, value in headers.items()],
     }
-    async with manager.run():
-        response_start, _ = await _call(manager, scope, body)
-        assert response_start["status"] == expected_status
-        assert manager._server_instances == {}
-        assert manager._session_owners == {}
+    with _created_transports() as transports:
+        async with manager.run():
+            response_start, _ = await _call(manager, scope, body)
+            assert response_start["status"] == expected_status
+            assert manager._server_instances == {}
+            assert manager._session_owners == {}
+            (transport,) = transports
+            assert transport.is_terminated
 
 
 @pytest.mark.anyio
diff --git a/tests/shared/test_streamable_http.py b/tests/shared/test_streamable_http.py
index 897f897c23..2e678a5c14 100644
--- a/tests/shared/test_streamable_http.py
+++ b/tests/shared/test_streamable_http.py
@@ -601,11 +601,12 @@ def test_streamable_http_transport_init_validation() -> None:
         StreamableHTTPServerTransport(mcp_session_id="test\n")
 
 
-@pytest.mark.parametrize("idle_timeout", [0, -1])
-def test_streamable_http_transport_rejects_non_positive_idle_timeout(idle_timeout: float) -> None:
-    """A transport's idle timeout must be a positive number of seconds; without one it never expires."""
-    with pytest.raises(ValueError, match="positive number of seconds"):
+@pytest.mark.parametrize("idle_timeout", [0, -1, float("inf"), float("nan")])
+def test_streamable_http_transport_rejects_invalid_idle_timeout(idle_timeout: float) -> None:
+    """A transport's idle timeout must be a positive, finite number of seconds; without one it never expires."""
+    with pytest.raises(ValueError) as exc_info:
         StreamableHTTPServerTransport(mcp_session_id="valid-id", idle_timeout=idle_timeout)
+    assert str(exc_info.value) == "idle_timeout must be a positive, finite number of seconds"
     assert StreamableHTTPServerTransport(mcp_session_id="valid-id").idle_scope is None
 
 
@@ -627,6 +628,50 @@ async def test_streamable_http_transport_creates_its_idle_scope_on_connect() ->
         await transport.terminate()
 
 
+async def _post_to_transport(transport: StreamableHTTPServerTransport, body: dict[str, Any]) -> int:
+    """POST `body` straight to `transport` in process, as a client that then holds the connection open,
+    and return the status it answered with."""
+    assert transport.mcp_session_id is not None
+    scope: Scope = {
+        "type": "http",
+        "method": "POST",
+        "path": "/mcp",
+        "query_string": b"",
+        "headers": [
+            (b"content-type", b"application/json"),
+            (b"accept", b"application/json, text/event-stream"),
+            (MCP_SESSION_ID_HEADER.encode(), transport.mcp_session_id.encode()),
+        ],
+    }
+    sent: list[Message] = []
+
+    async def send(message: Message) -> None:
+        sent.append(message)
+
+    request_body, incoming = anyio.create_memory_object_stream[Message](1)
+    async with request_body, incoming:
+        await request_body.send({"type": "http.request", "body": json.dumps(body).encode(), "more_body": False})
+        with anyio.fail_after(5):
+            await transport.handle_request(scope, incoming.receive, send)
+    return next(message["status"] for message in sent if message["type"] == "http.response.start")
+
+
+@pytest.mark.anyio
+async def test_transport_whose_idle_period_ran_out_answers_as_terminated() -> None:
+    """Once the idle scope has fired, a request that still reaches the transport is answered 404 and the
+    transport is terminated, instead of being dispatched into the message loop the host is leaving."""
+    transport = StreamableHTTPServerTransport(mcp_session_id="valid-id", idle_timeout=5)
+    ping = {"jsonrpc": "2.0", "id": 1, "method": "ping"}
+    async with transport.connect():
+        assert transport.idle_scope is not None
+        # Exactly what the scope's deadline passing does.
+        transport.idle_scope.cancel()
+
+        assert await _post_to_transport(transport, ping) == 404
+        assert transport.is_terminated
+        assert await _post_to_transport(transport, ping) == 404
+
+
 @pytest.mark.anyio
 async def test_transport_reports_stream_closure_when_host_exits_without_terminating(
     caplog: pytest.LogCaptureFixture,

From ab91d9c05f620168292324f429b17c178b79d1d3 Mon Sep 17 00:00:00 2001
From: Max Isbey <224885523+maxisbey@users.noreply.github.com>
Date: Wed, 26 Aug 2026 16:37:37 +0000
Subject: [PATCH 5/5] Report a full session table as a server error and hold
 the creation lock for admission only

Review follow-ups. The 503 for a request that would exceed
`max_sessions` now carries JSON-RPC code -32603 like the subscription
limit does, instead of -32600, which is what an unknown session ID
answers with; a client that reconnects on the latter should not treat a
full server the same way.

The session-creation lock only needs to make the limit check and the
registration atomic, so it is now held for that admission step alone and
the opening request (whose body the request-size limit has already
buffered) is served outside it. The new-session path is split into
`_admit_session` and `_serve_opening_request` accordingly. A test pins
that a client which never finishes sending its opening request does not
hold up another client's session.
---
 src/mcp/server/streamable_http_manager.py    | 161 ++++++++++---------
 tests/server/test_streamable_http_manager.py |  37 ++++-
 2 files changed, 124 insertions(+), 74 deletions(-)

diff --git a/src/mcp/server/streamable_http_manager.py b/src/mcp/server/streamable_http_manager.py
index e84be74bb7..a7efac3fbc 100644
--- a/src/mcp/server/streamable_http_manager.py
+++ b/src/mcp/server/streamable_http_manager.py
@@ -11,7 +11,7 @@
 
 import anyio
 from anyio.abc import TaskStatus
-from mcp_types import DEFAULT_NEGOTIATED_VERSION, INVALID_REQUEST, ErrorData, JSONRPCError
+from mcp_types import DEFAULT_NEGOTIATED_VERSION, INTERNAL_ERROR, INVALID_REQUEST, ErrorData, JSONRPCError
 from mcp_types.version import HANDSHAKE_PROTOCOL_VERSIONS
 from starlette.requests import Request
 from starlette.responses import Response
@@ -287,76 +287,17 @@ async def _handle_stateful_request(self, scope: Scope, receive: Receive, send: S
             return
 
         if request_mcp_session_id is None:
-            # New session case
-            logger.debug("Creating new transport")
+            # New session case. Admission (the session limit and registration)
+            # is decided under the lock; the request itself is served outside
+            # it, so one client that is slow to send its opening request does
+            # not hold up the others.
             async with self._session_creation_lock:
-                if self.max_sessions is not None and len(self._server_instances) >= self.max_sessions:
-                    logger.warning("Refusing to open a new session: %d sessions are already open", self.max_sessions)
-                    await _error_response("Too many open sessions", 503)(scope, receive, send)
-                    return
-
-                new_session_id = uuid4().hex
-                http_transport = StreamableHTTPServerTransport(
-                    mcp_session_id=new_session_id,
-                    is_json_response_enabled=self.json_response,
-                    event_store=self.event_store,  # May be None (no resumability)
-                    security_settings=self.security_settings,
-                    retry_interval=self.retry_interval,
-                    idle_timeout=self.session_idle_timeout,
-                )
-
-                async def run_server(*, task_status: TaskStatus[None] = anyio.TASK_STATUS_IGNORED) -> None:
-                    async with http_transport.connect() as streams:
-                        read_stream, write_stream = streams
-                        task_status.started()
-                        try:
-                            # The transport cancels its idle scope once no request
-                            # has been in flight for `session_idle_timeout`; that
-                            # ends the loop and execution continues after the
-                            # `with` block. Without a timeout there is nothing to fire.
-                            idle_scope = http_transport.idle_scope
-                            if idle_scope is None:
-                                idle_scope = anyio.CancelScope()
-                            with idle_scope:
-                                # Drive via `serve_loop` (not `Server.run()`) so the
-                                # manager's already-entered lifespan is reused
-                                # rather than re-entered per session.
-                                await serve_loop(
-                                    self.app,
-                                    read_stream,
-                                    write_stream,
-                                    lifespan_state=self._lifespan_state,
-                                    session_id=http_transport.mcp_session_id,
-                                )
-
-                            if idle_scope.cancelled_caught:
-                                logger.info(f"Session {new_session_id} idle timeout")
-                        except Exception:
-                            logger.exception(f"Session {new_session_id} crashed")
-                        finally:
-                            # However the session ended (client DELETE, idle
-                            # timeout, crash), discard it.
-                            await self._discard_session(new_session_id, http_transport)
-
-                if requestor is not None:
-                    self._session_owners[new_session_id] = requestor
-                self._server_instances[new_session_id] = http_transport
-                logger.info(f"Created new transport with session ID: {new_session_id}")
-
-                # Without a session ID only an initialize request can succeed,
-                # so if this one is refused, fails or is cancelled (or the
-                # session's server task cannot even be started) nothing was
-                # established: discard the session again rather than keep it
-                # (and its server task) around.
-                established = False
-                try:
-                    assert self._task_group is not None
-                    await self._task_group.start(run_server)
-                    status = await _send_and_report_status(http_transport.handle_request, scope, receive, send)
-                    established = status is not None and status < 400
-                finally:
-                    if not established:  # pragma: no branch
-                        await self._discard_session(new_session_id, http_transport)
+                http_transport = self._admit_session(requestor)
+            if http_transport is None:
+                logger.warning("Refusing to open a new session: %d sessions are already open", self.max_sessions)
+                await _error_response("Too many open sessions", 503, INTERNAL_ERROR)(scope, receive, send)
+                return
+            await self._serve_opening_request(http_transport, scope, receive, send)
         else:
             # Unknown or expired session ID - return 404 per MCP spec
             # TODO(L62): Align error code once spec clarifies
@@ -364,6 +305,82 @@ async def run_server(*, task_status: TaskStatus[None] = anyio.TASK_STATUS_IGNORE
             logger.info(f"Rejected request with unknown or expired session ID: {request_mcp_session_id[:64]}")
             await _error_response("Session not found", 404)(scope, receive, send)
 
+    def _admit_session(self, requestor: AuthorizationContext | None) -> StreamableHTTPServerTransport | None:
+        """Register a new session for `requestor` and return its transport, or None at the session limit."""
+        if self.max_sessions is not None and len(self._server_instances) >= self.max_sessions:
+            return None
+        http_transport = StreamableHTTPServerTransport(
+            mcp_session_id=uuid4().hex,
+            is_json_response_enabled=self.json_response,
+            event_store=self.event_store,  # May be None (no resumability)
+            security_settings=self.security_settings,
+            retry_interval=self.retry_interval,
+            idle_timeout=self.session_idle_timeout,
+        )
+        session_id = http_transport.mcp_session_id
+        assert session_id is not None
+        if requestor is not None:
+            self._session_owners[session_id] = requestor
+        self._server_instances[session_id] = http_transport
+        logger.info(f"Created new transport with session ID: {session_id}")
+        return http_transport
+
+    async def _serve_opening_request(
+        self, http_transport: StreamableHTTPServerTransport, scope: Scope, receive: Receive, send: Send
+    ) -> None:
+        """Start the session's server task and let its transport answer the request that opens it.
+
+        Without a session ID only an initialize request can succeed, so if this
+        one is refused, fails or is cancelled (or the session's server task
+        cannot even be started) nothing was established: the session is
+        discarded again rather than kept (with its server task) around.
+        """
+        session_id = http_transport.mcp_session_id
+        assert session_id is not None
+
+        async def run_server(*, task_status: TaskStatus[None] = anyio.TASK_STATUS_IGNORED) -> None:
+            async with http_transport.connect() as streams:
+                read_stream, write_stream = streams
+                task_status.started()
+                try:
+                    # The transport cancels its idle scope once no request
+                    # has been in flight for `session_idle_timeout`; that
+                    # ends the loop and execution continues after the
+                    # `with` block. Without a timeout there is nothing to fire.
+                    idle_scope = http_transport.idle_scope
+                    if idle_scope is None:
+                        idle_scope = anyio.CancelScope()
+                    with idle_scope:
+                        # Drive via `serve_loop` (not `Server.run()`) so the
+                        # manager's already-entered lifespan is reused
+                        # rather than re-entered per session.
+                        await serve_loop(
+                            self.app,
+                            read_stream,
+                            write_stream,
+                            lifespan_state=self._lifespan_state,
+                            session_id=session_id,
+                        )
+
+                    if idle_scope.cancelled_caught:
+                        logger.info(f"Session {session_id} idle timeout")
+                except Exception:
+                    logger.exception(f"Session {session_id} crashed")
+                finally:
+                    # However the session ended (client DELETE, idle
+                    # timeout, crash), discard it.
+                    await self._discard_session(session_id, http_transport)
+
+        established = False
+        try:
+            assert self._task_group is not None
+            await self._task_group.start(run_server)
+            status = await _send_and_report_status(http_transport.handle_request, scope, receive, send)
+            established = status is not None and status < 400
+        finally:
+            if not established:  # pragma: no branch
+                await self._discard_session(session_id, http_transport)
+
     async def _discard_session(self, session_id: str, transport: StreamableHTTPServerTransport) -> None:
         """Stop tracking the session and make sure its transport refuses anything that still reaches it.
 
@@ -378,9 +395,9 @@ async def _discard_session(self, session_id: str, transport: StreamableHTTPServe
                 await transport.terminate()
 
 
-def _error_response(message: str, status_code: int) -> Response:
+def _error_response(message: str, status_code: int, code: int = INVALID_REQUEST) -> Response:
     """A JSON-RPC error body (no request id) with the given HTTP status."""
-    body = JSONRPCError(jsonrpc="2.0", id=None, error=ErrorData(code=INVALID_REQUEST, message=message))
+    body = JSONRPCError(jsonrpc="2.0", id=None, error=ErrorData(code=code, message=message))
     return Response(
         body.model_dump_json(by_alias=True, exclude_unset=True), status_code=status_code, media_type="application/json"
     )
diff --git a/tests/server/test_streamable_http_manager.py b/tests/server/test_streamable_http_manager.py
index d281ebd62e..1b8f9264f1 100644
--- a/tests/server/test_streamable_http_manager.py
+++ b/tests/server/test_streamable_http_manager.py
@@ -5,13 +5,14 @@
 import math
 from collections.abc import AsyncIterator, Iterator
 from contextlib import asynccontextmanager, contextmanager
-from typing import Any
+from typing import Any, cast
 from unittest.mock import AsyncMock, patch
 
 import anyio
 import httpx2
 import pytest
 from mcp_types import (
+    INTERNAL_ERROR,
     INVALID_REQUEST,
     CallToolRequestParams,
     CallToolResult,
@@ -846,7 +847,7 @@ async def test_new_session_is_refused_at_max_sessions() -> None:
         assert json.loads(response_body) == {
             "jsonrpc": "2.0",
             "id": None,
-            "error": {"code": INVALID_REQUEST, "message": "Too many open sessions"},
+            "error": {"code": INTERNAL_ERROR, "message": "Too many open sessions"},
         }
         assert list(manager._server_instances) == [first]
 
@@ -855,6 +856,38 @@ async def test_new_session_is_refused_at_max_sessions() -> None:
         assert list(manager._server_instances) == [second]
 
 
+@pytest.mark.anyio
+async def test_client_that_is_slow_to_send_its_opening_request_does_not_hold_up_others() -> None:
+    """While one client has yet to finish sending the request that would open its session, another
+    client can still open one."""
+    manager = StreamableHTTPSessionManager(app=Server("test-slow-open"))
+    body_awaited = anyio.Event()
+
+    async def stall() -> None:
+        # This client has sent its headers but never finishes sending the body.
+        body_awaited.set()
+        await anyio.sleep_forever()
+
+    async def discard(message: Message) -> None: ...
+
+    slow_client = anyio.CancelScope()
+
+    async def open_slowly() -> None:
+        with slow_client:
+            await manager.handle_request(_request_scope(), cast(Receive, stall), discard)
+
+    session_id: str | None = None
+    async with manager.run():
+        async with anyio.create_task_group() as tg:
+            tg.start_soon(open_slowly)
+            with anyio.fail_after(5):
+                await body_awaited.wait()
+                session_id = await _open_session(manager, None)
+            slow_client.cancel()
+        assert session_id is not None
+        assert list(manager._server_instances) == [session_id]
+
+
 def test_max_sessions_defaults_to_ten_thousand() -> None:
     """A manager holds at most 10 000 concurrent stateful sessions unless configured otherwise."""
     manager = StreamableHTTPSessionManager(app=Server("test"))