test: add an in-process streaming ASGI transport and cover server-initiated requests over streamable HTTP

Author: maxisbey

tests/interaction/README.md

@@ -27,7 +27,9 @@ The whole suite is in-memory and event-driven; it runs in about a second.
   SDK's own deliberate output.
 - **No sleeps, no real I/O.** Concurrency is coordinated with `anyio.Event`; every wait that
   could hang is bounded by `anyio.fail_after(5)`. The streamable HTTP tests drive the Starlette
-  app in-process through `httpx.ASGITransport` — no sockets, threads, or subprocesses anywhere.
+  app in-process through the suite's streaming ASGI bridge (`transports/_bridge.py`), which
+  delivers each response chunk as the server produces it — full duplex, but still no sockets,
+  threads, or subprocesses anywhere.
 
 ## Layout
 

tests/interaction/_requirements.py

@@ -1722,7 +1722,8 @@ def __post_init__(self) -> None:
         source=f"{SPEC_BASE_URL}/basic/transports#streamable-http",
         behavior=(
             "A server-to-client message that is not related to an in-flight request is routed to the "
-            "standalone GET stream; a client that never opened one does not receive it."
+            "standalone GET stream and delivered to the client listening on it, not to any request's "
+            "own stream."
         ),
         transports=("streamable-http",),
     ),
@@ -1732,11 +1733,6 @@ def __post_init__(self) -> None:
             "A server-initiated request nested inside an in-flight call round-trips over stateful streamable HTTP."
         ),
         transports=("streamable-http",),
-        deferred=(
-            "The in-process ASGI client buffers each response in full, which deadlocks on a "
-            "server-to-client request nested inside a still-open call. Covered over a real socket by "
-            "tests/shared/test_streamable_http.py."
-        ),
     ),
     "transport:streamable-http:resumability": Requirement(
         source=f"{SPEC_BASE_URL}/basic/transports#streamable-http",

tests/interaction/test_coverage.py

@@ -26,6 +26,9 @@
 # Anything listed here is exempt from the every-test-has-a-requirement check.
 _HARNESS_SELF_TESTS = {
     "tests.interaction.lowlevel.test_wire.test_recording_read_stream_ends_iteration_when_the_sender_closes",
+    "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/_bridge.py

@@ -0,0 +1,154 @@
+"""An in-process, full-duplex HTTP transport for driving ASGI applications from httpx.
+
+`httpx.ASGITransport` runs the application to completion and only then hands the buffered
+response to the caller, so a server that streams its response — the streamable HTTP transport's
+SSE responses — can never converse with the client mid-request: a server-initiated request
+nested inside a still-open call deadlocks. `StreamingASGITransport` removes that limitation by
+running the application as a background task and forwarding every `http.response.body` chunk to
+the client the moment it is sent. Everything happens on the one event loop: no sockets, no
+threads, no sleeps, no extra dependencies.
+
+The behavioural contract, pinned by `test_bridge.py`:
+
+- The request body is buffered before the application is invoked (MCP requests are small JSON
+  documents); the response streams chunk by chunk.
+- Closing the response — or the whole client — delivers `http.disconnect` to the application,
+  exactly as a real server sees when its peer goes away.
+- An exception the application raises before sending `http.response.start` fails the originating
+  request with that same exception. After the response has started, a failure is visible to the
+  client only through the response itself (status code, truncated body) — the same signal a real
+  server over a real socket would give.
+
+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).
+"""
+
+import math
+from collections.abc import AsyncIterator
+from types import TracebackType
+
+import anyio
+import anyio.abc
+import httpx
+from anyio.streams.memory import MemoryObjectReceiveStream
+from starlette.types import ASGIApp, Message, Scope
+
+
+class _StreamingResponseBody(httpx.AsyncByteStream):
+    """A response body that yields chunks as the application produces them.
+
+    Closing it tells the application the client has gone away (`http.disconnect`), mirroring a
+    peer that drops the connection mid-response.
+    """
+
+    def __init__(self, chunks: MemoryObjectReceiveStream[bytes], client_disconnected: anyio.Event) -> None:
+        self._chunks = chunks
+        self._client_disconnected = client_disconnected
+
+    async def __aiter__(self) -> AsyncIterator[bytes]:
+        async for chunk in self._chunks:
+            yield chunk
+
+    async def aclose(self) -> None:
+        self._client_disconnected.set()
+        await self._chunks.aclose()
+
+
+class StreamingASGITransport(httpx.AsyncBaseTransport):
+    """Drive an ASGI application in-process, streaming each response as it is produced."""
+
+    _task_group: anyio.abc.TaskGroup
+
+    def __init__(self, app: ASGIApp) -> None:
+        self._app = app
+
+    async def __aenter__(self) -> "StreamingASGITransport":
+        self._task_group = anyio.create_task_group()
+        await self._task_group.__aenter__()
+        return self
+
+    async def __aexit__(
+        self,
+        exc_type: type[BaseException] | None = None,
+        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()
+        await self._task_group.__aexit__(exc_type, exc_value, traceback)
+
+    async def handle_async_request(self, request: httpx.Request) -> httpx.Response:
+        assert isinstance(request.stream, httpx.AsyncByteStream)
+        request_body = b"".join([chunk async for chunk in request.stream])
+
+        scope: Scope = {
+            "type": "http",
+            "asgi": {"version": "3.0"},
+            "http_version": "1.1",
+            "method": request.method,
+            "scheme": request.url.scheme,
+            "path": request.url.path,
+            "raw_path": request.url.raw_path.split(b"?", maxsplit=1)[0],
+            "query_string": request.url.query,
+            "root_path": "",
+            "headers": [(name.lower(), value) for name, value in request.headers.raw],
+            "server": (request.url.host, request.url.port),
+            "client": ("127.0.0.1", 1234),
+        }
+
+        request_delivered = False
+        client_disconnected = anyio.Event()
+        response_started = anyio.Event()
+        response_status = 0
+        response_headers: list[tuple[bytes, bytes]] = []
+        application_error: Exception | None = None
+        chunk_writer, chunk_reader = anyio.create_memory_object_stream[bytes](math.inf)
+
+        async def receive_request() -> Message:
+            nonlocal request_delivered
+            if not request_delivered:
+                request_delivered = True
+                return {"type": "http.request", "body": request_body, "more_body": False}
+            await client_disconnected.wait()
+            return {"type": "http.disconnect"}
+
+        async def send_response(message: Message) -> None:
+            nonlocal response_status, response_headers
+            if message["type"] == "http.response.start":
+                response_status = message["status"]
+                response_headers = list(message.get("headers", []))
+                response_started.set()
+                return
+            assert message["type"] == "http.response.body"
+            body: bytes = message.get("body", b"")
+            if body:
+                await chunk_writer.send(body)
+            if not message.get("more_body", False):
+                await chunk_writer.aclose()
+
+        async def run_application() -> None:
+            nonlocal application_error
+            try:
+                await self._app(scope, receive_request, send_response)
+            except Exception as exc:  # The bridge is the application's outermost boundary: a crash
+                # must fail the originating request (or show up in the already-started response),
+                # never tear down the task group shared with every other in-flight request.
+                application_error = exc
+            finally:
+                response_started.set()
+                await chunk_writer.aclose()
+
+        self._task_group.start_soon(run_application)
+        await response_started.wait()
+        if application_error is not None:
+            # No response will be built, so close the reader the response body would have owned.
+            await chunk_reader.aclose()
+            raise application_error
+        return httpx.Response(
+            status_code=response_status,
+            headers=response_headers,
+            stream=_StreamingResponseBody(chunk_reader, client_disconnected),
+            request=request,
+        )

tests/interaction/transports/test_bridge.py

@@ -0,0 +1,71 @@
+"""Contract tests for the suite's streaming ASGI bridge.
+
+These pin what `StreamingASGITransport` itself guarantees — chunk-by-chunk delivery, disconnect
+propagation, and failure handling — against minimal hand-written ASGI applications, so the MCP
+transport tests built on top of it never have to wonder what the harness provides. They are
+harness self-tests, not interaction-model tests, and are exempted from the requirement-coverage
+contract in `test_coverage.py`.
+"""
+
+import anyio
+import httpx
+import pytest
+from starlette.types import Message, Receive, Scope, Send
+
+from tests.interaction.transports._bridge import StreamingASGITransport
+
+pytestmark = pytest.mark.anyio
+
+
+async def test_response_chunks_arrive_as_the_application_sends_them() -> None:
+    """Each body chunk is delivered as sent, empty chunks are skipped, and the stream ends with the application."""
+
+    async def chunked_app(scope: Scope, receive: Receive, send: Send) -> None:
+        assert scope["type"] == "http"
+        assert (await receive())["type"] == "http.request"
+        await send({"type": "http.response.start", "status": 200, "headers": [(b"content-type", b"text/plain")]})
+        await send({"type": "http.response.body", "body": b"first", "more_body": True})
+        await send({"type": "http.response.body", "body": b"", "more_body": True})
+        await send({"type": "http.response.body", "body": b"second", "more_body": False})
+
+    async with httpx.AsyncClient(transport=StreamingASGITransport(chunked_app), base_url="http://bridge") as http:
+        async with http.stream("GET", "/chunks") as response:
+            with anyio.fail_after(5):
+                chunks = [chunk async for chunk in response.aiter_raw()]
+
+    assert response.status_code == 200
+    assert response.headers["content-type"] == "text/plain"
+    assert chunks == [b"first", b"second"]
+
+
+async def test_closing_the_response_delivers_a_disconnect_to_the_application() -> None:
+    """A client that closes the response early is seen by the application as an http.disconnect."""
+    seen_after_request: list[Message] = []
+    disconnect_seen = anyio.Event()
+
+    async def waiting_app(scope: Scope, receive: Receive, send: Send) -> None:
+        assert scope["type"] == "http"
+        assert (await receive())["type"] == "http.request"
+        await send({"type": "http.response.start", "status": 200, "headers": []})
+        seen_after_request.append(await receive())
+        disconnect_seen.set()
+
+    async with httpx.AsyncClient(transport=StreamingASGITransport(waiting_app), base_url="http://bridge") as http:
+        async with http.stream("GET", "/wait") as response:
+            assert response.status_code == 200
+        # Leaving the stream block closes the response while the application is still mid-response.
+        with anyio.fail_after(5):
+            await disconnect_seen.wait()
+
+    assert seen_after_request == [{"type": "http.disconnect"}]
+
+
+async def test_an_application_failure_before_the_response_starts_fails_the_request() -> None:
+    """An exception raised before http.response.start reaches the caller as that same exception."""
+
+    async def broken_app(scope: Scope, receive: Receive, send: Send) -> None:
+        raise RuntimeError("the demo application is broken")
+
+    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")