test: add streamable HTTP hosting, resumability, and client transport conformance tests

Author: maxisbey

src/mcp/client/streamable_http.py

@@ -210,7 +210,7 @@ async def handle_get_stream(self, client: httpx.AsyncClient, read_stream_writer:
                     # Stream ended normally (server closed) - reset attempt counter
                     attempt = 0
 
-            except Exception:  # pragma: lax no cover
+            except Exception:
                 logger.debug("GET stream error", exc_info=True)
                 attempt += 1
 
@@ -492,17 +492,17 @@ async def handle_request_async():
 
     async def terminate_session(self, client: httpx.AsyncClient) -> None:
         """Terminate the session by sending a DELETE request."""
-        if not self.session_id:  # pragma: lax no cover
-            return
+        if not self.session_id:
+            return  # pragma: no cover
 
         try:
             headers = self._prepare_headers()
             response = await client.delete(self.url, headers=headers)
 
-            if response.status_code == 405:  # pragma: lax no cover
+            if response.status_code == 405:
                 logger.debug("Server does not allow session termination")
-            elif response.status_code not in (200, 204):  # pragma: lax no cover
-                logger.warning(f"Session termination failed: {response.status_code}")
+            elif response.status_code not in (200, 204):
+                logger.warning(f"Session termination failed: {response.status_code}")  # pragma: no cover
         except Exception as exc:  # pragma: no cover
             logger.warning(f"Session termination failed: {exc}")
 

src/mcp/server/mcpserver/context.py

@@ -237,7 +237,7 @@ async def close_sse_stream(self) -> None:
             This is a no-op if not using StreamableHTTP transport with event_store.
             The callback is only available when event_store is configured.
         """
-        if self._request_context and self._request_context.close_sse_stream:  # pragma: no cover
+        if self._request_context and self._request_context.close_sse_stream:  # pragma: no branch
             await self._request_context.close_sse_stream()
 
     async def close_standalone_sse_stream(self) -> None:

src/mcp/server/streamable_http.py

@@ -173,7 +173,7 @@ async def run_stateless_server(*, task_status: TaskStatus[None] = anyio.TASK_STA
                         self.app.create_initialization_options(),
                         stateless=True,
                     )
-                except Exception:  # pragma: no cover
+                except Exception:  # pragma: lax no cover
                     logger.exception("Stateless session crashed")
 
         # Assert task group is not None for type checking

src/mcp/server/streamable_http_manager.py

@@ -40,19 +40,19 @@ def __init__(self, settings: TransportSecuritySettings | None = None):
         # If not specified, disable DNS rebinding protection by default for backwards compatibility
         self.settings = settings or TransportSecuritySettings(enable_dns_rebinding_protection=False)
 
-    def _validate_host(self, host: str | None) -> bool:  # pragma: no cover
+    def _validate_host(self, host: str | None) -> bool:
         """Validate the Host header against allowed values."""
-        if not host:
+        if not host:  # pragma: no cover
             logger.warning("Missing Host header in request")
             return False
 
         # Check exact match first
-        if host in self.settings.allowed_hosts:
+        if host in self.settings.allowed_hosts:  # pragma: no cover
             return True
 
         # Check wildcard port patterns
         for allowed in self.settings.allowed_hosts:
-            if allowed.endswith(":*"):
+            if allowed.endswith(":*"):  # pragma: no branch
                 # Extract base host from pattern
                 base_host = allowed[:-2]
                 # Check if the actual host starts with base host and has a port
@@ -62,19 +62,19 @@ def _validate_host(self, host: str | None) -> bool:  # pragma: no cover
         logger.warning(f"Invalid Host header: {host}")
         return False
 
-    def _validate_origin(self, origin: str | None) -> bool:  # pragma: no cover
+    def _validate_origin(self, origin: str | None) -> bool:
         """Validate the Origin header against allowed values."""
         # Origin can be absent for same-origin requests
-        if not origin:
+        if not origin:  # pragma: no cover
             return True
 
         # Check exact match first
-        if origin in self.settings.allowed_origins:
+        if origin in self.settings.allowed_origins:  # pragma: no cover
             return True
 
         # Check wildcard port patterns
         for allowed in self.settings.allowed_origins:
-            if allowed.endswith(":*"):
+            if allowed.endswith(":*"):  # pragma: no branch
                 # Extract base origin from pattern
                 base_origin = allowed[:-2]
                 # Check if the actual origin starts with base origin and has a port
@@ -103,14 +103,14 @@ async def validate_request(self, request: Request, is_post: bool = False) -> Res
         if not self.settings.enable_dns_rebinding_protection:
             return None
 
-        # Validate Host header  # pragma: no cover
-        host = request.headers.get("host")  # pragma: no cover
-        if not self._validate_host(host):  # pragma: no cover
-            return Response("Invalid Host header", status_code=421)  # pragma: no cover
+        # Validate Host header
+        host = request.headers.get("host")
+        if not self._validate_host(host):
+            return Response("Invalid Host header", status_code=421)
 
-        # Validate Origin header  # pragma: no cover
-        origin = request.headers.get("origin")  # pragma: no cover
-        if not self._validate_origin(origin):  # pragma: no cover
-            return Response("Invalid Origin header", status_code=403)  # pragma: no cover
+        # Validate Origin header
+        origin = request.headers.get("origin")
+        if not self._validate_origin(origin):
+            return Response("Invalid Origin header", status_code=403)
 
-        return None  # pragma: no cover
+        return None

src/mcp/server/transport_security.py

@@ -62,6 +62,12 @@ stream pair), the bare-`ClientSession` lifecycle tests, the real-clock timeout t
 machinery is transport-independent and must not race transport latency), and everything under
 `transports/`, which pins behaviour only observable on that transport.
 
+A transport conformance test in `transports/` speaks raw `httpx` against the mounted ASGI app
+**only** when its assertion is about HTTP semantics that `Client` cannot observe — status codes,
+response headers, SSE event fields, which stream a message travels on. Any other behaviour is
+asserted through a `Client`, connected to the mounted app via `client_via_http(http)` so several
+clients can share one session manager.
+
 ## The requirements manifest
 
 `_requirements.py` maps every behaviour the suite covers to the reason it must hold:

tests/interaction/README.md

@@ -9,11 +9,12 @@
 
 import gc
 import warnings
-from collections.abc import AsyncIterator
+from collections.abc import AsyncIterator, Awaitable, Callable, Iterable
 from contextlib import AbstractAsyncContextManager, asynccontextmanager
 from typing import Protocol
 
 import httpx
+from httpx_sse import ServerSentEvent, aconnect_sse
 from starlette.applications import Starlette
 from starlette.requests import Request
 from starlette.responses import Response
@@ -26,12 +27,29 @@
 from mcp.server import Server
 from mcp.server.mcpserver import MCPServer
 from mcp.server.sse import SseServerTransport
+from mcp.server.streamable_http import EventStore
+from mcp.server.streamable_http_manager import StreamableHTTPSessionManager
 from mcp.server.transport_security import TransportSecuritySettings
-from mcp.types import Implementation
+from mcp.types import (
+    LATEST_PROTOCOL_VERSION,
+    ClientCapabilities,
+    Implementation,
+    InitializeRequestParams,
+    JSONRPCMessage,
+    JSONRPCRequest,
+    JSONRPCResponse,
+    jsonrpc_message_adapter,
+)
 from tests.interaction.transports._bridge 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"
+BASE_URL = "http://127.0.0.1:8000"
+
+# DNS-rebinding protection validates Host/Origin headers against a real network attack that cannot
+# exist for an in-process ASGI app, so the in-process factories disable it; tests that exercise the
+# protection itself pass explicit settings (or transport_security=None to get the localhost
+# auto-enable behaviour).
+NO_DNS_REBINDING_PROTECTION = TransportSecuritySettings(enable_dns_rebinding_protection=False)
 
 
 class Connect(Protocol):
@@ -86,6 +104,8 @@ async def connect_over_streamable_http(
     *,
     stateless_http: bool = False,
     json_response: bool = False,
+    event_store: EventStore | None = None,
+    retry_interval: int | None = None,
     read_timeout_seconds: float | None = None,
     sampling_callback: SamplingFnT | None = None,
     list_roots_callback: ListRootsFnT | None = None,
@@ -98,19 +118,19 @@ async def connect_over_streamable_http(
 
     With the defaults this is the matrix leg (stateful sessions, SSE responses); the
     transport-specific tests pass `stateless_http` or `json_response` to select the other
-    server modes.
+    server modes, and the resumability tests pass an `event_store` (with `retry_interval=0` so
+    the client's reconnection wait is a no-op).
     """
-    # DNS-rebinding protection validates Host/Origin headers against a real network attack that
-    # cannot exist for an in-process ASGI app; leaving it on would also pull the origin-validation
-    # branch (deliberately uncovered in src) into coverage.
     app = server.streamable_http_app(
         stateless_http=stateless_http,
         json_response=json_response,
-        transport_security=TransportSecuritySettings(enable_dns_rebinding_protection=False),
+        event_store=event_store,
+        retry_interval=retry_interval,
+        transport_security=NO_DNS_REBINDING_PROTECTION,
     )
     async with server.session_manager.run():
-        async with httpx.AsyncClient(transport=StreamingASGITransport(app), base_url=_BASE_URL) as http_client:
-            transport = streamable_http_client(f"{_BASE_URL}/mcp", http_client=http_client)
+        async with httpx.AsyncClient(transport=StreamingASGITransport(app), base_url=BASE_URL) as http_client:
+            transport = streamable_http_client(f"{BASE_URL}/mcp", http_client=http_client)
             async with Client(
                 transport,
                 read_timeout_seconds=read_timeout_seconds,
@@ -124,6 +144,139 @@ async def connect_over_streamable_http(
                 yield client
 
 
+@asynccontextmanager
+async def mounted_app(
+    server: Server | MCPServer,
+    *,
+    stateless_http: bool = False,
+    event_store: EventStore | None = None,
+    retry_interval: int | None = None,
+    transport_security: TransportSecuritySettings | None = NO_DNS_REBINDING_PROTECTION,
+    on_request: Callable[[httpx.Request], Awaitable[None]] | None = None,
+    headers: dict[str, str] | None = None,
+) -> AsyncIterator[tuple[httpx.AsyncClient, StreamableHTTPSessionManager]]:
+    """Mount the server's streamable HTTP app on the in-process bridge and yield an httpx client.
+
+    Yields the httpx client (rooted at the in-process origin) and the live session manager. Tests
+    use this in two ways: for raw-httpx assertions (status codes, headers, SSE bytes) the test
+    speaks HTTP through the yielded client directly; for client-driven assertions the test wraps
+    that client in `client_via_http(http)`, which lets several `Client`s share the one mounted
+    session manager. `on_request` records every outgoing HTTP request before it leaves the
+    yielded client.
+
+    DNS-rebinding protection is disabled by default; pass explicit settings (or `None` for the
+    localhost auto-enable behaviour) to test the protection itself.
+    """
+    app = server.streamable_http_app(
+        stateless_http=stateless_http,
+        event_store=event_store,
+        retry_interval=retry_interval,
+        transport_security=transport_security,
+    )
+    event_hooks = {"request": [on_request]} if on_request is not None else None
+    async with server.session_manager.run():
+        async with httpx.AsyncClient(
+            transport=StreamingASGITransport(app), base_url=BASE_URL, event_hooks=event_hooks, headers=headers
+        ) as http_client:
+            yield http_client, server.session_manager
+
+
+@asynccontextmanager
+async def client_via_http(
+    http_client: httpx.AsyncClient,
+    *,
+    logging_callback: LoggingFnT | None = None,
+    message_handler: MessageHandlerFnT | None = None,
+    elicitation_callback: ElicitationFnT | None = None,
+) -> AsyncIterator[Client]:
+    """Connect a `Client` over an already-mounted streamable HTTP app.
+
+    Use with `mounted_app(...)` so several `Client`s share the one session manager, or so a
+    client-driven assertion can sit alongside raw-httpx assertions in the same test. The
+    underlying `httpx.AsyncClient` is left open when the `Client` exits.
+    """
+    transport = streamable_http_client(f"{BASE_URL}/mcp", http_client=http_client)
+    async with Client(
+        transport,
+        logging_callback=logging_callback,
+        message_handler=message_handler,
+        elicitation_callback=elicitation_callback,
+    ) as client:
+        yield client
+
+
+def parse_sse_messages(events: Iterable[ServerSentEvent]) -> list[JSONRPCMessage]:
+    """Decode SSE events into JSON-RPC messages, skipping priming events that carry no data."""
+    return [jsonrpc_message_adapter.validate_json(event.data) for event in events if event.data]
+
+
+async def post_jsonrpc(
+    http: httpx.AsyncClient, body: dict[str, object], *, session_id: str | None = None
+) -> tuple[httpx.Response, list[JSONRPCMessage]]:
+    """POST a JSON-RPC body and read its SSE response stream to completion.
+
+    Returns the HTTP response (for header/status assertions) and the parsed JSON-RPC messages
+    that arrived on the response's SSE stream. Only meaningful for requests the server answers
+    with `text/event-stream`; for error responses or 202 notification acknowledgements, use
+    `httpx.AsyncClient.post` directly and assert on the response.
+    """
+    async with aconnect_sse(http, "POST", "/mcp", json=body, headers=base_headers(session_id=session_id)) as source:
+        events = [event async for event in source.aiter_sse()]
+    return source.response, parse_sse_messages(events)
+
+
+def base_headers(*, session_id: str | None = None) -> dict[str, str]:
+    """Standard request headers for raw-httpx streamable-HTTP tests.
+
+    Every well-formed request carries these (Accept covering both response representations,
+    Content-Type for POST bodies, MCP-Protocol-Version at the latest revision, and the session
+    ID once one exists), so a test that wants to assert a specific rejection only varies the one
+    header under test.
+    """
+    headers = {
+        "accept": "application/json, text/event-stream",
+        "content-type": "application/json",
+        "mcp-protocol-version": LATEST_PROTOCOL_VERSION,
+    }
+    if session_id is not None:
+        headers["mcp-session-id"] = session_id
+    return headers
+
+
+def initialize_body(request_id: int = 1) -> dict[str, object]:
+    """A wire-level initialize JSON-RPC request body, exactly as an SDK client would send it."""
+    params = InitializeRequestParams(
+        protocol_version=LATEST_PROTOCOL_VERSION,
+        capabilities=ClientCapabilities(),
+        client_info=Implementation(name="raw", version="0.0.0"),
+    )
+    return JSONRPCRequest(
+        jsonrpc="2.0", id=request_id, method="initialize", params=params.model_dump(by_alias=True, exclude_none=True)
+    ).model_dump(by_alias=True, exclude_none=True)
+
+
+async def initialize_via_http(http: httpx.AsyncClient) -> str:
+    """Perform the initialize handshake over a raw `httpx.AsyncClient` and return the session ID.
+
+    Validates the SSE response and sends the `notifications/initialized` follow-up, so the server
+    is fully ready for subsequent feature requests when this returns.
+    """
+    async with aconnect_sse(http, "POST", "/mcp", json=initialize_body(), headers=base_headers()) as source:
+        assert source.response.status_code == 200
+        # An event-store-backed server opens the stream with a priming event (empty data); skip it.
+        events = [event async for event in source.aiter_sse() if event.data]
+    assert len(events) == 1
+    assert JSONRPCResponse.model_validate_json(events[0].data).id == 1
+    session_id = source.response.headers["mcp-session-id"]
+    initialized = await http.post(
+        "/mcp",
+        json={"jsonrpc": "2.0", "method": "notifications/initialized"},
+        headers=base_headers(session_id=session_id),
+    )
+    assert initialized.status_code == 202
+    return session_id
+
+
 def build_sse_app(server: Server | MCPServer) -> tuple[Starlette, SseServerTransport]:
     """Mount a server on a Starlette app exposing the legacy SSE transport at /sse and /messages/.
 
@@ -175,13 +328,13 @@ def httpx_client_factory(
         # bridge must let the application drain rather than cancelling at close.
         return httpx.AsyncClient(
             transport=StreamingASGITransport(app, cancel_on_close=False),
-            base_url=_BASE_URL,
+            base_url=BASE_URL,
             headers=headers,
             timeout=timeout,
             auth=auth,
         )
 
-    transport = sse_client(f"{_BASE_URL}/sse", httpx_client_factory=httpx_client_factory)
+    transport = sse_client(f"{BASE_URL}/sse", httpx_client_factory=httpx_client_factory)
     try:
         async with Client(
             transport,