test: run the interaction suite over the legacy SSE transport in-process

Author: maxisbey

src/mcp/server/sse.py

@@ -116,15 +116,15 @@ def __init__(self, endpoint: str, security_settings: TransportSecuritySettings |
         logger.debug(f"SseServerTransport initialized with endpoint: {endpoint}")
 
     @asynccontextmanager
-    async def connect_sse(self, scope: Scope, receive: Receive, send: Send):  # pragma: no cover
-        if scope["type"] != "http":
+    async def connect_sse(self, scope: Scope, receive: Receive, send: Send):
+        if scope["type"] != "http":  # pragma: no cover
             logger.error("connect_sse received non-HTTP request")
             raise ValueError("connect_sse can only handle HTTP requests")
 
         # Validate request headers for DNS rebinding protection
         request = Request(scope, receive)
         error_response = await self._security.validate_request(request, is_post=False)
-        if error_response:
+        if error_response:  # pragma: no cover
             await error_response(scope, receive, send)
             raise ValueError("Request validation failed")
 
@@ -190,13 +190,13 @@ async def response_wrapper(scope: Scope, receive: Receive, send: Send):
             logger.debug("Yielding read and write streams")
             yield (read_stream, write_stream)
 
-    async def handle_post_message(self, scope: Scope, receive: Receive, send: Send) -> None:  # pragma: no cover
+    async def handle_post_message(self, scope: Scope, receive: Receive, send: Send) -> None:
         logger.debug("Handling POST message")
         request = Request(scope, receive)
 
         # Validate request headers for DNS rebinding protection
         error_response = await self._security.validate_request(request, is_post=True)
-        if error_response:
+        if error_response:  # pragma: no cover
             return await error_response(scope, receive, send)
 
         session_id_param = request.query_params.get("session_id")
@@ -225,7 +225,7 @@ async def handle_post_message(self, scope: Scope, receive: Receive, send: Send)
         try:
             message = types.jsonrpc_message_adapter.validate_json(body, by_name=False)
             logger.debug(f"Validated client message: {message}")
-        except ValidationError as err:
+        except ValidationError as err:  # pragma: no cover
             logger.exception("Failed to parse message")
             response = Response("Could not parse message", status_code=400)
             await response(scope, receive, send)

tests/interaction/_connect.py

@@ -1,24 +1,31 @@
 """Transport-parametrized connection factories for the interaction suite.
 
 The `connect` fixture (see conftest.py) hands tests one of these factories so the same test body
-runs over the in-memory transport and over streamable HTTP without naming either: the factory is a
-drop-in replacement for constructing `Client(server, ...)` and yields the connected client. The
-streamable HTTP factory drives the server's real Starlette app through the in-process streaming
-bridge, so the full HTTP framing layer (session ids, SSE encoding, session management) runs with
-no sockets, threads, or subprocesses.
+runs over each transport without naming any of them: the factory is a drop-in replacement for
+constructing `Client(server, ...)` and yields the connected client. The HTTP factories drive the
+server's real Starlette app through the in-process streaming bridge, so the full transport layer
+(session ids, SSE encoding, session management) runs with no sockets, threads, or subprocesses.
 """
 
+import gc
+import warnings
 from collections.abc import AsyncIterator
 from contextlib import AbstractAsyncContextManager, asynccontextmanager
 from typing import Protocol
 
 import httpx
+from starlette.applications import Starlette
+from starlette.requests import Request
+from starlette.responses import Response
+from starlette.routing import Mount, Route
 
 from mcp.client.client import Client
 from mcp.client.session import ElicitationFnT, ListRootsFnT, LoggingFnT, MessageHandlerFnT, SamplingFnT
+from mcp.client.sse import sse_client
 from mcp.client.streamable_http import streamable_http_client
 from mcp.server import Server
 from mcp.server.mcpserver import MCPServer
+from mcp.server.sse import SseServerTransport
 from mcp.server.transport_security import TransportSecuritySettings
 from mcp.types import Implementation
 from tests.interaction.transports._bridge import StreamingASGITransport
@@ -115,3 +122,84 @@ async def connect_over_streamable_http(
                 elicitation_callback=elicitation_callback,
             ) as client:
                 yield client
+
+
+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/.
+
+    `MCPServer.sse_app()` exists but does not expose the underlying `SseServerTransport`, which
+    the SSE-specific tests need; building the app explicitly here gives both server flavours the
+    same routing while keeping that handle.
+    """
+    sse = SseServerTransport(
+        "/messages/", security_settings=TransportSecuritySettings(enable_dns_rebinding_protection=False)
+    )
+    lowlevel = server._lowlevel_server if isinstance(server, MCPServer) else server
+
+    async def handle_sse(request: Request) -> Response:
+        async with sse.connect_sse(request.scope, request.receive, request._send) as (read, write):
+            await lowlevel.run(read, write, lowlevel.create_initialization_options())
+        return Response()
+
+    app = Starlette(
+        routes=[
+            Route("/sse", endpoint=handle_sse, methods=["GET"]),
+            Mount("/messages/", app=sse.handle_post_message),
+        ],
+    )
+    return app, sse
+
+
+@asynccontextmanager
+async def connect_over_sse(
+    server: Server | MCPServer,
+    *,
+    read_timeout_seconds: float | None = None,
+    sampling_callback: SamplingFnT | None = None,
+    list_roots_callback: ListRootsFnT | None = None,
+    logging_callback: LoggingFnT | None = None,
+    message_handler: MessageHandlerFnT | None = None,
+    client_info: Implementation | None = None,
+    elicitation_callback: ElicitationFnT | None = None,
+) -> AsyncIterator[Client]:
+    """Yield a Client connected to the server's legacy SSE transport, entirely in process."""
+    app, _ = build_sse_app(server)
+
+    def httpx_client_factory(
+        headers: dict[str, str] | None = None,
+        timeout: httpx.Timeout | None = None,
+        auth: httpx.Auth | None = None,
+    ) -> httpx.AsyncClient:
+        # The SSE server transport's connect_sse runs the entire MCP session inside the GET
+        # request and only releases its streams after that request observes a disconnect, so the
+        # 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,
+            headers=headers,
+            timeout=timeout,
+            auth=auth,
+        )
+
+    transport = sse_client(f"{_BASE_URL}/sse", httpx_client_factory=httpx_client_factory)
+    try:
+        async with Client(
+            transport,
+            read_timeout_seconds=read_timeout_seconds,
+            sampling_callback=sampling_callback,
+            list_roots_callback=list_roots_callback,
+            logging_callback=logging_callback,
+            message_handler=message_handler,
+            client_info=client_info,
+            elicitation_callback=elicitation_callback,
+        ) as client:
+            yield client
+    finally:
+        # SseServerTransport.connect_sse hands its internal SSE-chunk receive stream to
+        # sse_starlette's EventSourceResponse, which never closes it when its task group is
+        # cancelled on disconnect (see notes/findings.md). Collect the orphan here so its
+        # ResourceWarning fires deterministically inside this fixture instead of at an
+        # arbitrary later GC.
+        with warnings.catch_warnings():
+            warnings.simplefilter("ignore", ResourceWarning)
+            gc.collect()

tests/interaction/_requirements.py

@@ -1759,10 +1759,23 @@ def __post_init__(self) -> None:
             "requests, with server messages delivered on the SSE stream."
         ),
         transports=("sse",),
-        deferred=(
-            "The legacy SSE transport is covered by tests/shared/test_sse.py; in-process coverage in this "
-            "suite arrives with the transport fixture work."
+    ),
+    "transport:sse:endpoint-event": Requirement(
+        source=f"{SPEC_BASE_URL}/basic/transports#backwards-compatibility",
+        behavior=(
+            "Opening the SSE stream delivers an `endpoint` event naming the message-POST URL and a fresh "
+            "session identifier; the server registers the session before the event is sent and releases it "
+            "when the stream disconnects."
+        ),
+        transports=("sse",),
+    ),
+    "transport:sse:post:session-routing": Requirement(
+        source="sdk",
+        behavior=(
+            "A POST to the SSE message endpoint that names no session id, a malformed session id, or an "
+            "unknown session id is rejected (400/400/404) instead of being forwarded."
         ),
+        transports=("sse",),
     ),
     "transport:stdio": Requirement(
         source=f"{SPEC_BASE_URL}/basic/transports#stdio",

tests/interaction/conftest.py

@@ -2,11 +2,12 @@
 
 import pytest
 
-from tests.interaction._connect import Connect, connect_in_memory, connect_over_streamable_http
+from tests.interaction._connect import Connect, connect_in_memory, connect_over_sse, connect_over_streamable_http
 
 _FACTORIES: dict[str, Connect] = {
     "in-memory": connect_in_memory,
     "streamable-http": connect_over_streamable_http,
+    "sse": connect_over_sse,
 }
 
 

tests/interaction/test_coverage.py

@@ -29,6 +29,7 @@
     "tests.interaction.transports.test_bridge.test_response_chunks_arrive_as_the_application_sends_them",
     "tests.interaction.transports.test_bridge.test_closing_the_response_delivers_a_disconnect_to_the_application",
     "tests.interaction.transports.test_bridge.test_an_application_failure_before_the_response_starts_fails_the_request",
+    "tests.interaction.transports.test_bridge.test_disabling_cancel_on_close_lets_the_application_finish_after_disconnect",
 }
 
 

tests/interaction/transports/_bridge.py

@@ -21,7 +21,8 @@
 
 The transport owns an anyio task group for the application tasks; it is opened and closed by
 `httpx.AsyncClient`'s own context manager, so use the client as a context manager (the suite
-always does).
+always does). Closing the transport cancels every running application task by default; set
+`cancel_on_close=False` to wait for the application's own disconnect handling instead.
 """
 
 import math
@@ -56,12 +57,19 @@ async def aclose(self) -> None:
 
 
 class StreamingASGITransport(httpx.AsyncBaseTransport):
-    """Drive an ASGI application in-process, streaming each response as it is produced."""
+    """Drive an ASGI application in-process, streaming each response as it is produced.
+
+    With `cancel_on_close` (the default), closing the transport cancels every application task
+    still running so harness teardown can never hang. Setting it to False makes the transport wait
+    for the application's own disconnect handling to complete instead, which is the path the legacy
+    SSE server transport relies on for resource cleanup.
+    """
 
     _task_group: anyio.abc.TaskGroup
 
-    def __init__(self, app: ASGIApp) -> None:
+    def __init__(self, app: ASGIApp, *, cancel_on_close: bool = True) -> None:
         self._app = app
+        self._cancel_on_close = cancel_on_close
 
     async def __aenter__(self) -> "StreamingASGITransport":
         self._task_group = anyio.create_task_group()
@@ -74,9 +82,11 @@ async def __aexit__(
         exc_value: BaseException | None = None,
         traceback: TracebackType | None = None,
     ) -> None:
-        # Any application task still running at this point is serving a client that no longer
-        # exists; cancel rather than wait so harness teardown can never hang.
-        self._task_group.cancel_scope.cancel()
+        # httpx closes every streamed response before closing the transport, so by now each
+        # application task has been delivered `http.disconnect`. Either cancel immediately, or wait
+        # for the application's own disconnect handling to unwind.
+        if self._cancel_on_close:
+            self._task_group.cancel_scope.cancel()
         await self._task_group.__aexit__(exc_type, exc_value, traceback)
 
     async def handle_async_request(self, request: httpx.Request) -> httpx.Response:

tests/interaction/transports/test_bridge.py

@@ -69,3 +69,24 @@ async def broken_app(scope: Scope, receive: Receive, send: Send) -> None:
     async with httpx.AsyncClient(transport=StreamingASGITransport(broken_app), base_url="http://bridge") as http:
         with pytest.raises(RuntimeError, match="the demo application is broken"):
             await http.get("/broken")
+
+
+async def test_disabling_cancel_on_close_lets_the_application_finish_after_disconnect() -> None:
+    """With cancel_on_close=False, an application that runs cleanup after seeing http.disconnect
+    completes that cleanup before the transport finishes closing."""
+    cleanup_ran = anyio.Event()
+
+    async def lingering_app(scope: Scope, receive: Receive, send: Send) -> None:
+        assert scope["type"] == "http"
+        await receive()
+        await send({"type": "http.response.start", "status": 200, "headers": []})
+        assert (await receive())["type"] == "http.disconnect"
+        cleanup_ran.set()
+
+    transport = StreamingASGITransport(lingering_app, cancel_on_close=False)
+    async with httpx.AsyncClient(transport=transport, base_url="http://bridge") as http:
+        with anyio.fail_after(5):
+            async with http.stream("GET", "/linger") as response:
+                assert response.status_code == 200
+            assert not cleanup_ran.is_set()
+    assert cleanup_ran.is_set()

tests/interaction/transports/test_sse.py

@@ -0,0 +1,98 @@
+"""Behaviour specific to the legacy HTTP+SSE transport, exercised entirely in process.
+
+Transport-agnostic behaviour is covered by the `connect`-fixture matrix, which runs the rest of
+the suite over this transport as well; this file pins only what is observable on the SSE wiring
+itself: the GET-then-POST connection lifecycle, the endpoint event, and how the message endpoint
+rejects requests it cannot route to a session. Every test drives the server's real Starlette app
+through the suite's streaming ASGI bridge.
+"""
+
+import gc
+import warnings
+from uuid import UUID, uuid4
+
+import anyio
+import httpx
+import pytest
+from inline_snapshot import snapshot
+
+from mcp.client.client import Client
+from mcp.client.sse import sse_client
+from mcp.server import Server
+from mcp.types import EmptyResult
+from tests.interaction._connect import build_sse_app
+from tests.interaction._requirements import requirement
+from tests.interaction.transports._bridge import StreamingASGITransport
+
+pytestmark = pytest.mark.anyio
+
+_BASE_URL = "http://127.0.0.1:8000"
+
+
+@requirement("transport:sse")
+@requirement("transport:sse:endpoint-event")
+async def test_endpoint_event_names_the_message_endpoint_with_a_fresh_session_id() -> None:
+    """Connecting opens a GET stream whose first event names the POST endpoint and a fresh
+    session id; messages POSTed there are answered on that stream, and disconnecting releases the
+    server's session entry."""
+    app, sse = build_sse_app(Server("legacy"))
+    captured_session_id: list[str] = []
+
+    def httpx_client_factory(
+        headers: dict[str, str] | None = None,
+        timeout: httpx.Timeout | None = None,
+        auth: httpx.Auth | None = None,
+    ) -> httpx.AsyncClient:
+        return httpx.AsyncClient(
+            transport=StreamingASGITransport(app, cancel_on_close=False),
+            base_url=_BASE_URL,
+            headers=headers,
+            timeout=timeout,
+            auth=auth,
+        )
+
+    transport = sse_client(
+        f"{_BASE_URL}/sse", httpx_client_factory=httpx_client_factory, on_session_created=captured_session_id.append
+    )
+    with anyio.fail_after(5):
+        async with Client(transport) as client:
+            assert len(captured_session_id) == 1
+            assert UUID(hex=captured_session_id[0]) in sse._read_stream_writers
+            assert await client.send_ping() == snapshot(EmptyResult())
+
+    assert sse._read_stream_writers == {}
+    # See connect_over_sse: collect the one stream sse_starlette never closes on disconnect.
+    with warnings.catch_warnings():
+        warnings.simplefilter("ignore", ResourceWarning)
+        gc.collect()
+
+
+@requirement("transport:sse:post:session-routing")
+async def test_post_without_a_session_id_is_rejected() -> None:
+    """A POST to the message endpoint with no session_id query parameter is answered 400."""
+    app, _ = build_sse_app(Server("legacy"))
+    async with httpx.AsyncClient(transport=StreamingASGITransport(app), base_url=_BASE_URL) as http:
+        response = await http.post("/messages/", json={"jsonrpc": "2.0", "method": "ping", "id": 1})
+    assert (response.status_code, response.text) == snapshot((400, "session_id is required"))
+
+
+@requirement("transport:sse:post:session-routing")
+async def test_post_with_a_malformed_session_id_is_rejected() -> None:
+    """A POST whose session_id query parameter is not a UUID is answered 400."""
+    app, _ = build_sse_app(Server("legacy"))
+    async with httpx.AsyncClient(transport=StreamingASGITransport(app), base_url=_BASE_URL) as http:
+        response = await http.post(
+            "/messages/", params={"session_id": "not-a-uuid"}, json={"jsonrpc": "2.0", "method": "ping", "id": 1}
+        )
+    assert (response.status_code, response.text) == snapshot((400, "Invalid session ID"))
+
+
+@requirement("transport:sse:post:session-routing")
+async def test_post_for_an_unknown_session_is_rejected() -> None:
+    """A POST naming a well-formed session_id that no SSE stream owns is answered 404."""
+    app, _ = build_sse_app(Server("legacy"))
+    async with httpx.AsyncClient(transport=StreamingASGITransport(app), base_url=_BASE_URL) as http:
+        response = await http.post(
+            "/messages/", params={"session_id": uuid4().hex}, json={"jsonrpc": "2.0", "method": "ping", "id": 1}
+        )
+    assert (response.status_code, response.text) == snapshot((404, "Could not find session"))