[v1.x] Expire idle Streamable HTTP sessions by default and cap concurrent sessions (#3426)

docs/server.md

@@ -1254,6 +1254,7 @@ The FastMCP server instance accessible via `ctx.fastmcp` provides access to serv
   - `mount_path`, `sse_path`, `streamable_http_path` - Transport paths
   - `stateless_http` - Whether the server operates in stateless mode
   - `max_request_body_size` - Maximum HTTP request body size in bytes (Streamable HTTP and SSE)
+  - `session_idle_timeout` and `max_sessions` - Streamable HTTP session expiry and session cap
   - And other configuration options
 
 ```python
@@ -1426,6 +1427,9 @@ messages, configure the smallest suitable byte limit:
 mcp = FastMCP("Large messages", max_request_body_size=8 * 1024 * 1024)
 ```
 
+Stateful sessions expire and are capped per process. See
+[Session lifetime and limits](#session-lifetime-and-limits) below.
+
 <!-- snippet-source examples/snippets/servers/streamable_config.py -->
 ```python
 """
@@ -1536,6 +1540,39 @@ The streamable HTTP transport supports:
 - JSON or SSE response formats
 - Better scalability for multi-node deployments
 
+#### Session lifetime and limits
+
+A stateful session does not live forever, and one process does not hold an unlimited number of
+them. Two settings control this. Both are keyword arguments on `FastMCP(...)`. `stateless_http=True`
+keeps no sessions, so neither applies there.
+
+| Setting | Default | What it does | What the client sees | Turn it off |
+|---|---|---|---|---|
+| `session_idle_timeout` | `1800` (30 min) | Closes a session that has had nothing in flight for that long. | `404 Session not found`. It has to `initialize` again. | `None` |
+| `max_sessions` | `10_000` | Refuses to open a session beyond that many. Existing sessions are untouched and nothing is evicted. | `503 Too many open sessions` with JSON-RPC code `-32603`. | `None` |
+
+What counts as "in flight":
+
+- An open `GET` stream. The SDK clients keep one open, so a connected client's session never
+  expires.
+- A request that is still being answered. A tool call that runs longer than the timeout is not
+  interrupted, and the countdown only starts once it finishes.
+- Nothing else. Between requests the clock runs. Any request on the session restarts it,
+  `ping` included. Once a session has expired, nothing revives it.
+
+A client that ends its session with `DELETE` frees it immediately. So does a client whose
+opening request was refused.
+
+```python
+mcp = FastMCP("My server", session_idle_timeout=None, max_sessions=50_000)
+```
+
+Both events show up in the server log. An expiry is `Session <id> idle timeout` at `INFO`. A
+refused open is `Refusing to open a new session: <n> sessions are already open` at `WARNING`.
+
+The limits are per process. With four workers the ceiling is four times `max_sessions`, and each
+worker expires its own sessions.
+
 #### CORS Configuration for Browser-Based Clients
 
 If you'd like your server to be accessible by browser-based MCP clients, you'll need to configure CORS headers. The `Mcp-Session-Id` header must be exposed for browser clients to access it:

src/mcp/server/fastmcp/server.py

@@ -62,7 +62,11 @@ from mcp.server.session import ServerSession, ServerSessionT
 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.transport_security import DEFAULT_MAX_REQUEST_BODY_SIZE, TransportSecuritySettings
 from mcp.shared.context import LifespanContextT, RequestContext, RequestT
 from mcp.types import Annotations, AnyFunction, ContentBlock, GetPromptResult, Icon, ToolAnnotations
@@ -108,6 +112,10 @@ class Settings(BaseSettings, Generic[LifespanResultT]):
     """Define if the server should create a new transport per request."""
     max_request_body_size: int
     """Maximum request body size in bytes for the Streamable HTTP endpoint and the SSE message endpoint."""
+    session_idle_timeout: float | None
+    """Seconds a stateful session may have no request in flight before it is closed. None disables expiry."""
+    max_sessions: int | None
+    """Maximum number of concurrent stateful sessions. None removes the limit."""
 
     # resource settings
     warn_on_duplicate_resources: bool
@@ -169,6 +177,8 @@ class FastMCP(Generic[LifespanResultT]):
         json_response: bool = False,
         stateless_http: bool = False,
         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,
         warn_on_duplicate_resources: bool = True,
         warn_on_duplicate_tools: bool = True,
         warn_on_duplicate_prompts: bool = True,
@@ -197,6 +207,8 @@ class FastMCP(Generic[LifespanResultT]):
             json_response=json_response,
             stateless_http=stateless_http,
             max_request_body_size=max_request_body_size,
+            session_idle_timeout=session_idle_timeout,
+            max_sessions=max_sessions,
             warn_on_duplicate_resources=warn_on_duplicate_resources,
             warn_on_duplicate_tools=warn_on_duplicate_tools,
             warn_on_duplicate_prompts=warn_on_duplicate_prompts,
@@ -966,6 +978,8 @@ class FastMCP(Generic[LifespanResultT]):
                 stateless=self.settings.stateless_http,  # Use the stateless setting
                 security_settings=self.settings.transport_security,
                 max_request_body_size=self.settings.max_request_body_size,
+                session_idle_timeout=self.settings.session_idle_timeout,
+                max_sessions=self.settings.max_sessions,
             )
 
         # Create the ASGI handler

src/mcp/server/streamable_http.py

@@ -9,6 +9,7 @@ responses, with streaming support for long-running operations.
 
 import json
 import logging
+import math
 import re
 from abc import ABC, abstractmethod
 from collections.abc import AsyncGenerator, Awaitable, Callable
@@ -150,6 +151,7 @@ class StreamableHTTPServerTransport:
         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.
@@ -167,12 +169,22 @@ class StreamableHTTPServerTransport:
                            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 waits on `idle_scope`
+                         (available once `connect()` has been entered) and ends 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, 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 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
@@ -188,8 +200,11 @@ class StreamableHTTPServerTransport:
         ] = {}
         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:
@@ -402,6 +417,32 @@ class StreamableHTTPServerTransport:
 
     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
+
+        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.
+        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
@@ -643,8 +684,9 @@ class StreamableHTTPServerTransport:
                 except Exception:
                     logger.exception("SSE response error")
                     await sse_stream_writer.aclose()
-                    await sse_stream_reader.aclose()
                     await self._clean_up_memory_streams(request_id)
+                finally:
+                    await sse_stream_reader.aclose()
 
         except Exception as err:
             logger.exception("Error handling POST request")
@@ -747,9 +789,10 @@ class StreamableHTTPServerTransport:
             await response(request.scope, request.receive, send)
         except Exception:
             logger.exception("Error in standalone SSE response")
+            await self._clean_up_memory_streams(GET_STREAM_KEY)
+        finally:
             await sse_stream_writer.aclose()
             await sse_stream_reader.aclose()
-            await self._clean_up_memory_streams(GET_STREAM_KEY)
 
     async def _handle_delete_request(self, request: Request, send: Send) -> None:  # pragma: no cover
         """Handle DELETE requests for explicit session termination."""
@@ -992,6 +1035,8 @@ class StreamableHTTPServerTransport:
         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
 

src/mcp/server/streamable_http_manager.py

@@ -4,15 +4,16 @@ from __future__ import annotations
 
 import contextlib
 import logging
+import math
 from collections.abc import AsyncIterator
-from typing import Any
+from typing import Any, Final
 from uuid import uuid4
 
 import anyio
 from anyio.abc import TaskStatus
 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.auth.middleware.bearer_auth import AuthenticatedUser, AuthorizationContext, authorization_context
 from mcp.server.lowlevel.server import Server as MCPServer
@@ -26,12 +27,24 @@ from mcp.server.transport_security import (
     RequestBodyLimitMiddleware,
     TransportSecuritySettings,
 )
-from mcp.types import INVALID_REQUEST, ErrorData, JSONRPCError
+from mcp.types import INTERNAL_ERROR, INVALID_REQUEST, ErrorData, JSONRPCError
 
-__all__ = ["DEFAULT_MAX_REQUEST_BODY_SIZE", "RequestBodyLimitMiddleware", "StreamableHTTPSessionManager"]
+__all__ = [
+    "DEFAULT_MAX_REQUEST_BODY_SIZE",
+    "DEFAULT_MAX_SESSIONS",
+    "DEFAULT_SESSION_IDLE_TIMEOUT",
+    "RequestBodyLimitMiddleware",
+    "StreamableHTTPSessionManager",
+]
 
 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)."""
+
+DEFAULT_MAX_SESSIONS: Final = 10_000
+"""Default maximum number of concurrent stateful Streamable HTTP sessions per session manager."""
+
 
 class StreamableHTTPSessionManager:
     """
@@ -44,7 +57,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
@@ -61,13 +74,18 @@ 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
+        session_idle_timeout: Idle timeout in seconds for stateful sessions. A session that has had no HTTP
-            receive no HTTP requests for this duration will be automatically terminated and removed. When
+            request in flight for this long (no request being served, no open GET stream) is terminated and
-            retry_interval is also configured, ensure the idle timeout comfortably exceeds the retry interval to
+            removed; its ID then answers 404 and the client has to initialize a new session. When retry_interval
-            avoid reaping sessions during normal SSE polling gaps. Default is None (no timeout). A value of 1800
+            is also configured, ensure the idle timeout comfortably exceeds the retry interval to avoid reaping
-            (30 minutes) is recommended for most deployments.
+            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.
+        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__(
@@ -78,15 +96,16 @@ class StreamableHTTPSessionManager:
         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,
+        max_sessions: int | None = DEFAULT_MAX_SESSIONS,
     ):
-        if session_idle_timeout is not None and session_idle_timeout <= 0:
+        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 number of seconds")
+            raise ValueError("session_idle_timeout must be a positive, finite 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")
+        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
@@ -96,6 +115,7 @@ class StreamableHTTPSessionManager:
         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)
@@ -224,16 +244,15 @@ class StreamableHTTPSessionManager:
                 except Exception:  # pragma: 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
+        try:
-        await self._task_group.start(run_stateless_server)
+            await self._task_group.start(run_stateless_server)
-
+            await http_transport.handle_request(scope, receive, send)
-        # Handle the HTTP request and return the response
+        finally:
-        await http_transport.handle_request(scope, receive, send)
+            with anyio.CancelScope(shield=True):
-
+                await http_transport.terminate()
-        # Terminate the transport after the request is handled
-        await http_transport.terminate()
 
     async def _handle_stateful_request(
         self,
@@ -265,100 +284,142 @@ class StreamableHTTPSessionManager:
                     "Rejecting request for session %s: credential does not match the one that created the session",
                     request_mcp_session_id[:64],
                 )
-                body = JSONRPCError(
+                await _error_response("Session not found", 404)(scope, receive, send)
-                    jsonrpc="2.0", id="server-error", error=ErrorData(code=INVALID_REQUEST, message="Session not found")
-                )
-                response = Response(
-                    body.model_dump_json(by_alias=True, exclude_none=True),
-                    status_code=404,
-                    media_type="application/json",
-                )
-                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:  # pragma: no cover
-                transport.idle_scope.deadline = anyio.current_time() + self.session_idle_timeout
             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.
+                await self._discard_session(request_mcp_session_id, transport)
             return
 
         if request_mcp_session_id is None:
-            # New session case
+            # New session case. Admission (the session limit and registration)
-            logger.debug("Creating new transport")
+            # 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:
-                new_session_id = uuid4().hex
+                http_transport = self._admit_session(requestor)
-                http_transport = StreamableHTTPServerTransport(
+            if http_transport is None:
-                    mcp_session_id=new_session_id,
+                logger.warning("Refusing to open a new session: %d sessions are already open", self.max_sessions)
-                    is_json_response_enabled=self.json_response,
+                await _error_response("Too many open sessions", 503, INTERNAL_ERROR)(scope, receive, send)
-                    event_store=self.event_store,  # May be None (no resumability)
+                return
-                    security_settings=self.security_settings,
+            await self._serve_opening_request(http_transport, scope, receive, send)
-                    retry_interval=self.retry_interval,
-                )
-
-                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
-                        task_status.started()
-                        try:
-                            # Use a cancel scope for idle timeout — when the
-                            # deadline passes the scope cancels app.run() and
-                            # execution continues after the ``with`` block.
-                            # Incoming requests push the deadline forward.
-                            idle_scope = anyio.CancelScope()
-                            if self.session_idle_timeout is not None:
-                                idle_scope.deadline = anyio.current_time() + self.session_idle_timeout
-                                http_transport.idle_scope = idle_scope
-
-                            with idle_scope:
-                                await self.app.run(
-                                    read_stream,
-                                    write_stream,
-                                    self.app.create_initialization_options(),
-                                    stateless=False,
-                                )
-
-                            if idle_scope.cancelled_caught:
-                                assert http_transport.mcp_session_id is not None
-                                logger.info(f"Session {http_transport.mcp_session_id} idle timeout")
-                                self._server_instances.pop(http_transport.mcp_session_id, None)
-                                self._session_owners.pop(http_transport.mcp_session_id, None)
-                                await http_transport.terminate()
-                        except Exception:
-                            logger.exception(f"Session {http_transport.mcp_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)
-
-                # 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)
         else:
             # Unknown or expired session ID - return 404 per MCP spec
-            body = JSONRPCError(
+            await _error_response("Session not found", 404)(scope, receive, send)
-                jsonrpc="2.0", id="server-error", error=ErrorData(code=INVALID_REQUEST, message="Session not found")
+
-            )
+    def _admit_session(self, requestor: AuthorizationContext | None) -> StreamableHTTPServerTransport | None:
-            response = Response(
+        """Register a new session for `requestor` and return its transport, or None at the session limit."""
-                body.model_dump_json(by_alias=True, exclude_none=True), status_code=404, media_type="application/json"
+        if self.max_sessions is not None and len(self._server_instances) >= self.max_sessions:
-            )
+            return None
-            await response(scope, receive, send)
+        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:
+                    async with anyio.create_task_group() as session_tg:
+                        if http_transport.idle_scope is not None:
+
+                            async def end_when_idle(idle_scope: anyio.CancelScope) -> None:
+                                # The transport cancels this scope once no request
+                                # has been in flight for `session_idle_timeout`.
+                                with idle_scope:
+                                    await anyio.sleep_forever()
+                                logger.info(f"Session {session_id} idle timeout")
+                                # Discarding the session closes the transport's
+                                # streams, so app.run() returns the way it does
+                                # after a client DELETE and the server's lifespan
+                                # for this session is torn down normally.
+                                await self._discard_session(session_id, http_transport)
+
+                            session_tg.start_soon(end_when_idle, http_transport.idle_scope)
+                        await self.app.run(
+                            read_stream,
+                            write_stream,
+                            self.app.create_initialization_options(),
+                            stateless=False,
+                        )
+                        # The session ended some other way; stop waiting for it to go idle.
+                        session_tg.cancel_scope.cancel()
+                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.
+
+        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, code: int = INVALID_REQUEST) -> Response:
+    """A JSON-RPC error body (no usable request id) with the given HTTP status."""
+    body = JSONRPCError(jsonrpc="2.0", id="server-error", error=ErrorData(code=code, message=message))
+    return Response(
+        body.model_dump_json(by_alias=True, exclude_none=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
+
+    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

tests/conftest.py

@@ -1,6 +1,27 @@
+from collections.abc import Iterator
+
 import pytest
+from sse_starlette.sse import AppStatus
 
 
 @pytest.fixture
 def anyio_backend():
     return "asyncio"
+
+
+@pytest.fixture(autouse=True)
+def reset_sse_starlette_exit_event() -> Iterator[None]:
+    """sse-starlette<2 caches a module-level anyio.Event on AppStatus. Clear it
+    around each test so it is never bound to a closed event loop: any test that
+    serves an SSE response in process would otherwise inherit the event a
+    previous test created on another loop. Clearing it afterwards matters too,
+    because later test modules fork uvicorn subprocesses on Linux and would
+    otherwise inherit a stale event."""
+
+    def clear() -> None:
+        if hasattr(AppStatus, "should_exit_event"):  # pragma: no cover
+            setattr(AppStatus, "should_exit_event", None)
+
+    clear()
+    yield
+    clear()

tests/server/fastmcp/test_server.py

@@ -1544,3 +1544,20 @@ async def test_sse_app_applies_the_configured_request_body_limit() -> None:
             headers={"Content-Type": "application/json"},
         )
     assert response.status_code == 413
+
+
+def test_streamable_http_app_passes_the_configured_session_limits_to_its_manager() -> None:
+    """SDK-defined: FastMCP forwards `session_idle_timeout` and `max_sessions` to the Streamable HTTP manager;
+    by default sessions expire after 30 idle minutes and one process holds at most 10 000 of them."""
+    default = FastMCP()
+    default.streamable_http_app()
+    assert (default.session_manager.session_idle_timeout, default.session_manager.max_sessions) == (30 * 60, 10_000)
+
+    tuned = FastMCP(session_idle_timeout=5, max_sessions=7)
+    assert (tuned.settings.session_idle_timeout, tuned.settings.max_sessions) == (5, 7)
+    tuned.streamable_http_app()
+    assert (tuned.session_manager.session_idle_timeout, tuned.session_manager.max_sessions) == (5, 7)
+
+    unbounded = FastMCP(session_idle_timeout=None, max_sessions=None)
+    unbounded.streamable_http_app()
+    assert (unbounded.session_manager.session_idle_timeout, unbounded.session_manager.max_sessions) == (None, None)

tests/server/test_sse_security.py

@@ -4,7 +4,6 @@ import logging
 import multiprocessing
 import re
 import socket
-from collections.abc import Iterator
 from typing import Any
 
 import anyio
@@ -29,23 +28,6 @@ logger = logging.getLogger(__name__)
 SERVER_NAME = "test_sse_security_server"
 
 
-@pytest.fixture(autouse=True)
-def reset_sse_starlette_exit_event() -> Iterator[None]:
-    """sse-starlette<2 caches a module-level anyio.Event on AppStatus; clear it
-    around each test so it is never bound to a closed event loop. Clearing it
-    afterwards matters too: later test modules fork uvicorn subprocesses on
-    Linux and would otherwise inherit a stale event."""
-    from sse_starlette.sse import AppStatus
-
-    def clear() -> None:
-        if hasattr(AppStatus, "should_exit_event"):  # pragma: no cover
-            setattr(AppStatus, "should_exit_event", None)
-
-    clear()
-    yield
-    clear()
-
-
 @pytest.fixture
 def server_port() -> int:
     with socket.socket() as s:

tests/server/test_streamable_http_manager.py

@@ -1,13 +1,17 @@
 """Tests for StreamableHTTPSessionManager."""
 
 import json
-from collections.abc import Iterator
+import logging
-from typing import Any
+import math
+from collections.abc import AsyncIterator, Iterator
+from contextlib import asynccontextmanager, contextmanager
+from typing import Any, cast
 from unittest.mock import AsyncMock, patch
 
 import anyio
+import anyio.lowlevel
 import pytest
-from starlette.types import Message, Scope
+from starlette.types import Message, Receive, Scope, Send
 
 from mcp.server import streamable_http_manager
 from mcp.server.auth.middleware.bearer_auth import AuthenticatedUser
@@ -16,9 +20,27 @@ from mcp.server.lowlevel import Server
 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,
 )
-from mcp.types import INVALID_REQUEST
+from mcp.types import INTERNAL_ERROR, INVALID_REQUEST, LATEST_PROTOCOL_VERSION, TextContent
+
+_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_PROTOCOL_VERSION,
+            "capabilities": {},
+            "clientInfo": {"name": "test", "version": "0"},
+        },
+    }
+).encode()
+"""A wire-level initialize request: the only request that may open a session."""
 
 
 @pytest.mark.anyio
@@ -285,19 +307,7 @@ async def test_stateless_requests_memory_cleanup():
     app = Server("test-stateless-real-cleanup")
     manager = StreamableHTTPSessionManager(app=app, stateless=True)
 
-    # Track created transport instances
+    with _created_transports() as created_transports:
-    created_transports: list[StreamableHTTPServerTransport] = []
-
-    # Patch StreamableHTTPServerTransport constructor to track instances
-
-    original_constructor = streamable_http_manager.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):
         async with manager.run():
             # Mock app.run to complete immediately
             app.run = AsyncMock(return_value=None)
@@ -390,81 +400,497 @@ async def test_unknown_session_id_returns_404():
         assert error_data["error"]["message"] == "Session not found"
 
 
+class _IdleTimeoutObserver(logging.Handler):
+    """Resolves `reaped` when the manager logs that a session's idle timeout fired."""
+
+    def __init__(self) -> None:
+        super().__init__()
+        self.reaped = anyio.Event()
+
+    def emit(self, record: logging.LogRecord) -> None:
+        if "idle timeout" in record.getMessage():
+            self.reaped.set()
+
+
+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 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
+
+
+@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 _open_event_stream(
+    manager: StreamableHTTPSessionManager, session_id: str
+) -> AsyncIterator[StreamableHTTPServerTransport]:
+    """Hold the session's standalone GET stream open while the context is open, the way a listening client
+    does, and close it (the client goes away) on exit. Yields the session's transport once the stream has
+    been answered."""
+    stream_opened = anyio.Event()
+    client_gone = anyio.Event()
+    sent_messages: list[Message] = []
+    request_delivered = False
+
+    async def send(message: Message) -> None:
+        sent_messages.append(message)
+        stream_opened.set()
+
+    async def receive() -> Message:
+        # A GET carries an empty body; after that the client just holds the
+        # stream open until it goes away.
+        nonlocal request_delivered
+        if not request_delivered:
+            request_delivered = True
+            return {"type": "http.request", "body": b"", "more_body": False}
+        await client_gone.wait()
+        return {"type": "http.disconnect"}
+
+    async with anyio.create_task_group() as tg:
+        tg.start_soon(manager.handle_request, _request_scope(session_id=session_id, method="GET"), receive, send)
+        with anyio.fail_after(5):
+            await stream_opened.wait()
+        assert (sent_messages[0]["type"], sent_messages[0]["status"]) == ("http.response.start", 200)
+        yield manager._server_instances[session_id]
+        client_gone.set()
+
+
 @pytest.mark.anyio
-async def test_idle_session_is_reaped():
+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)
 
     async with manager.run():
-        sent_messages: list[Message] = []
+        session_id = await _open_session(manager, None)
 
-        async def mock_send(message: Message):
+        # Wait for the 50ms idle timeout to fire and the session to be unregistered. Re-requesting
-            sent_messages.append(message)
+        # the session to poll for the 404 would push its idle deadline forward and keep it alive.
-
+        with anyio.fail_after(5):
-        scope = {
+            await observer.reaped.wait()
-            "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
-
-        assert session_id is not None, "Session ID not found in response headers"
-
-        # Wait for the 50ms idle timeout to fire and cleanup to complete
-        await anyio.sleep(0.1)
 
         # Verify via public API: old session ID now returns 404
-        response_messages: list[Message] = []
+        assert await _request_session(manager, session_id, None) == 404
 
-        async def capture_send(message: Message):
-            response_messages.append(message)
 
-        scope_with_session = {
+@pytest.mark.anyio
-            "type": "http",
+async def test_expired_session_runs_the_server_lifespan_to_completion(
-            "method": "POST",
+    caplog: pytest.LogCaptureFixture, request: pytest.FixtureRequest
-            "path": "/mcp",
+) -> None:
-            "headers": [
+    """When a session expires, the server's lifespan for it is torn down the way it is after a DELETE:
-                (b"content-type", b"application/json"),
+    cleanup that awaits runs to completion instead of being cancelled part way."""
-                (b"mcp-session-id", session_id.encode()),
+    torn_down = anyio.Event()
-            ],
+    teardown: list[str] = []
+
+    @asynccontextmanager
+    async def lifespan(server: Server[Any, Any]) -> AsyncIterator[dict[str, Any]]:
+        try:
+            yield {}
+        finally:
+            try:
+                await anyio.lowlevel.checkpoint()  # stands in for cleanup that has to await
+                teardown.append("completed")
+            finally:
+                torn_down.set()
+
+    manager = StreamableHTTPSessionManager(app=Server("test-lifespan", lifespan=lifespan), session_idle_timeout=0.05)
+    observer = _observe_idle_timeout(caplog, request)
+
+    async with manager.run():
+        await _open_session(manager, None)
+        with anyio.fail_after(5):
+            await observer.reaped.wait()
+            await torn_down.wait()
+        assert teardown == ["completed"]
+
+
+@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()
+    app = Server("test-in-flight")
+
+    @app.call_tool()
+    async def handle_call_tool(name: str, arguments: dict[str, Any]) -> list[TextContent]:
+        tool_started.set()
+        await release_tool.wait()
+        return [TextContent(type="text", text="done")]
+
+    manager = StreamableHTTPSessionManager(app=app, session_idle_timeout=30)
+    observer = _observe_idle_timeout(caplog, request)
+
+    async with manager.run():
+        session_id = await _open_session(manager, None)
+        transport = manager._server_instances[session_id]
+        call_tool_body = json.dumps(
+            {"jsonrpc": "2.0", "id": 2, "method": "tools/call", "params": {"name": "slow", "arguments": {}}}
+        ).encode()
+        responses: list[tuple[Message, bytes]] = []
+
+        async def call_tool() -> None:
+            responses.append(await _call(manager, _request_scope(session_id=session_id), call_tool_body))
+
+        async with anyio.create_task_group() as tg:
+            tg.start_soon(call_tool)
+            with anyio.fail_after(5):
+                await tool_started.wait()
+            # 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()
+
+        response_start, response_body = responses[0]
+        assert response_start["status"] == 200
+        assert b'"done"' in response_body
+
+        # 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
+        assert transport.is_terminated
+        assert await _request_session(manager, session_id, None) == 404
+
+
+@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=30)
+    observer = _observe_idle_timeout(caplog, request)
+
+    async with manager.run():
+        session_id = await _open_session(manager, None)
+
+        async with _open_event_stream(manager, session_id) as transport:
+            # The stream has been answered, so it is in flight: 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 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
+        assert await _request_session(manager, session_id, None) == 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=30)
+    observer = _observe_idle_timeout(caplog, request)
+
+    async with manager.run():
+        session_id = await _open_session(manager, None)
+
+        async with _open_event_stream(manager, session_id) as transport:
+            assert transport.idle_scope is not None and transport.idle_scope.deadline == math.inf
+            ping = b'{"jsonrpc": "2.0", "id": 2, "method": "ping"}'
+            response_start, _ = await _call(manager, _request_scope(session_id=session_id), ping)
+            assert response_start["status"] == 200
+            # The ping has completed (in-flight bookkeeping included, since `_call` only returns once the
+            # manager has), 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
+        assert await _request_session(manager, session_id, None) == 404
+
+
+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
+
+
+@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
+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
+@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"), session_idle_timeout=session_idle_timeout)
+    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": "server-error",
+            "error": {"code": INVALID_REQUEST, "message": "Session not found"},
         }
 
-        await manager.handle_request(scope_with_session, mock_receive, capture_send)
 
-        response_start = next(
+@pytest.mark.anyio
-            (msg for msg in response_messages if msg["type"] == "http.response.start"),
+async def test_opening_request_that_fails_leaves_no_session() -> None:
-            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."""
-        assert response_start is not None
+    manager = StreamableHTTPSessionManager(app=Server("test-failed-open"))
-        assert response_start["status"] == 404
+    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
 
 
-def test_session_idle_timeout_rejects_non_positive():
+@pytest.mark.anyio
-    with pytest.raises(ValueError, match="positive number"):
+async def test_opening_request_that_is_cancelled_leaves_no_session() -> None:
-        StreamableHTTPSessionManager(app=Server("test"), session_idle_timeout=-1)
+    """If the request that would open a session is cancelled while it is being served (the client went
-    with pytest.raises(ValueError, match="positive number"):
+    away), the provisional session is discarded rather than left registered."""
-        StreamableHTTPSessionManager(app=Server("test"), session_idle_timeout=0)
+    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)
+
+    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):
+                    await entered.wait()
+                assert len(manager._server_instances) == 1
+                opening_request.cancel()
+            assert manager._server_instances == {}
+            assert manager._session_owners == {}
+            (transport,) = transports
+            assert transport.is_terminated
 
 
-def test_session_idle_timeout_rejects_stateless():
+@pytest.mark.anyio
-    with pytest.raises(RuntimeError, match="not supported in stateless"):
+async def test_opening_request_whose_session_task_cannot_start_leaves_no_session() -> None:
-        StreamableHTTPSessionManager(app=Server("test"), session_idle_timeout=30, stateless=True)
+    """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
+@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()],
+    }
+    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
+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": "server-error",
+            "error": {"code": INTERNAL_ERROR, "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]
+
+
+@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()
+
+    slow_client = anyio.CancelScope()
+
+    async def open_slowly() -> None:
+        with slow_client:
+            # Nothing is ever sent back to this client, so any `send` will do.
+            await manager.handle_request(_request_scope(), cast(Receive, stall), AsyncMock())
+
+    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"))
+    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:
@@ -494,19 +920,33 @@ def _request_scope(
     return scope
 
 
-async def _open_session(manager: StreamableHTTPSessionManager, user: AuthenticatedUser | None) -> str:
+async def _call(manager: StreamableHTTPSessionManager, scope: Scope, body: bytes = b"") -> tuple[Message, bytes]:
-    """Create a new session as `user` and return its session ID."""
+    """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:
+    async def receive() -> Message:
-        return {"type": "http.request", "body": b"", "more_body": False}
+        # Deliver the body once, then block like a client holding the connection
-
+        # open; a streaming response ends when the server closes it.
-    await manager.handle_request(_request_scope(user=user), mock_receive, mock_send)
+        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
+
+
+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()
 
@@ -515,26 +955,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] = []
+    response_start, _ = await _call(manager, _request_scope(session_id=session_id, user=user, method=method))
-
-    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")
     return response_start["status"]
 
 
 @pytest.fixture
 async def manager_with_live_session():
-    """A running manager around a real `Server`. Sessions remain registered until
+    """A running manager around a real `Server`. Sessions are opened with a real initialize and stay
-    `manager.run()` exits because `Server.run` blocks waiting for an initialize message."""
+    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

tests/shared/test_streamable_http.py

@@ -9,6 +9,7 @@ import multiprocessing
 import socket
 import time
 from collections.abc import Generator
+from concurrent.futures import ThreadPoolExecutor
 from datetime import timedelta
 from typing import Any
 from unittest.mock import MagicMock
@@ -23,6 +24,7 @@ from pydantic import AnyUrl
 from starlette.applications import Starlette
 from starlette.requests import Request
 from starlette.routing import Mount
+from starlette.types import Message, Scope
 
 import mcp.types as types
 from mcp.client.session import ClientSession
@@ -719,6 +721,77 @@ def test_streamable_http_transport_init_validation():
         StreamableHTTPServerTransport(mcp_session_id="test\n")
 
 
+@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
+
+
+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()
+
+
+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
+
+
 def test_session_termination(basic_server: None, basic_server_url: str):
     """Test session termination via DELETE and subsequent request handling."""
     response = requests.post(
@@ -756,7 +829,7 @@ def test_session_termination(basic_server: None, basic_server_url: str):
         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"
 
 
 def test_response(basic_server: None, basic_server_url: str):