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: `. 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: `. 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 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: ` (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.py b/src/mcp/server/streamable_http.py index 1a4e9939a4..416dd9e2b4 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, 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 @@ -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,32 @@ 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 + + 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 @@ -793,7 +834,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 +1036,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 e9a7d9629b..a7efac3fbc 100644 --- a/src/mcp/server/streamable_http_manager.py +++ b/src/mcp/server/streamable_http_manager.py @@ -4,17 +4,18 @@ import contextlib import logging +import math from collections.abc import AsyncIterator -from typing import TYPE_CHECKING, Any +from typing import TYPE_CHECKING, Any, Final from uuid import uuid4 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 -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 @@ -34,6 +35,12 @@ 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: """Manages StreamableHTTP sessions with optional resumability via event store. @@ -45,7 +52,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,13 +69,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 - 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. + 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__( @@ -79,15 +91,16 @@ 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, + 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 stateless and session_idle_timeout is not None: - raise RuntimeError("session_idle_timeout is not supported in stateless mode") + 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: + raise ValueError("max_sessions must be a positive number of sessions or None") self.app = app self.event_store = event_store @@ -97,6 +110,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) @@ -234,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.""" @@ -263,109 +276,145 @@ 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") - # 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 + # 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 - 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: - 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, - ) - - 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 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 - - 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: - 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) + 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 # 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 _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. + + 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 request id) with the given HTTP status.""" + 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" + ) + + +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: 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/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..1b8f9264f1 100644 --- a/tests/server/test_streamable_http_manager.py +++ b/tests/server/test_streamable_http_manager.py @@ -2,15 +2,28 @@ import json import logging -from collections.abc import Iterator -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 httpx2 import pytest -from mcp_types import INVALID_REQUEST, ListToolsResult, PaginatedRequestParams -from starlette.types import Message, Scope +from mcp_types import ( + INTERNAL_ERROR, + 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 ASGIApp, Message, Receive, Scope, Send from mcp import Client from mcp.client.streamable_http import streamable_http_client @@ -18,7 +31,32 @@ 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_MAX_SESSIONS, + 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"} + +_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 @@ -286,19 +324,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] = [] @@ -420,50 +446,57 @@ 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) +@contextmanager +def _created_transports() -> Iterator[list[StreamableHTTPServerTransport]]: + """Collect every transport a session manager creates while the context is open.""" + created: list[StreamableHTTPServerTransport] = [] - scope = { - "type": "http", - "method": "POST", - "path": "/mcp", - "headers": [(b"content-type", b"application/json")], - } + def create(*args: Any, **kwargs: Any) -> StreamableHTTPServerTransport: + transport = StreamableHTTPServerTransport(*args, **kwargs) + created.append(transport) + return transport - async def mock_receive(): - return {"type": "http.request", "body": b"", "more_body": False} + with patch.object(streamable_http_manager, "StreamableHTTPServerTransport", side_effect=create): + yield created - 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 +@asynccontextmanager +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 + + +@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,41 +504,402 @@ async def mock_receive(): await observer.reaped.wait() # 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 = { - "type": "http", - "method": "POST", - "path": "/mcp", - "headers": [ - (b"content-type", b"application/json"), - (b"mcp-session-id", session_id.encode()), - ], +@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=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, + "method": "tools/call", + "params": {"name": "slow", "arguments": {}}, } + responses: list[httpx2.Response] = [] - await manager.handle_request(scope_with_session, mock_receive, capture_send) + async def call_tool() -> None: + responses.append(await http.post("/mcp", json=call_tool_body, headers=session_headers)) - response_start = next( - (msg for msg in response_messages if msg["type"] == "http.response.start"), - None, - ) - assert response_start is not None + 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() + + 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 + 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 +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 _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 + # 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 + 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 + + +@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) + session_post_served = anyio.Event() + + 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 + 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 + 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: + """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": None, + "error": {"code": INVALID_REQUEST, "message": "Session not found"}, + } -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.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")) + 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 +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() -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) + 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 + + +@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 +@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": None, + "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() + + 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")) + 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: @@ -535,19 +929,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 +964,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..2e678a5c14 100644 --- a/tests/shared/test_streamable_http.py +++ b/tests/shared/test_streamable_http.py @@ -7,8 +7,10 @@ from __future__ import annotations as _annotations import json +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 @@ -599,6 +601,97 @@ def test_streamable_http_transport_init_validation() -> None: 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 + + +@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 +732,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 +1141,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 +1204,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