test: update SSE tests to use StreamingASGITransport instead of uvicorn for server testing

src/mcp/server/sse.py

@@ -190,6 +190,7 @@ async def response_wrapper(scope: Scope, receive: Receive, send: Send):
                 )
                 await read_stream_writer.aclose()
                 await write_stream_reader.aclose()
+                await sse_stream_reader.aclose()
                 logging.debug(f"Client session disconnected {session_id}")
 
             logger.debug("Starting SSE response task")

src/mcp/server/streaming_asgi_transport.py

@@ -9,6 +9,7 @@
 """
 
 import typing
+from collections.abc import Awaitable, Callable
 from typing import Any, cast
 
 import anyio
@@ -65,6 +66,8 @@ async def handle_async_request(
     ) -> Response:
         assert isinstance(request.stream, AsyncByteStream)
 
+        disconnect_event = anyio.Event()
+
         # ASGI scope.
         scope = {
             "type": "http",
@@ -97,11 +100,17 @@ async def handle_async_request(
         content_send_channel, content_receive_channel = anyio.create_memory_object_stream[bytes](100)
 
         # ASGI callables.
+        async def send_disconnect() -> None:
+            disconnect_event.set()
+
         async def receive() -> dict[str, Any]:
             nonlocal request_complete
 
+            if disconnect_event.is_set():
+                return {"type": "http.disconnect"}
+
             if request_complete:
-                await response_complete.wait()
+                await disconnect_event.wait()
                 return {"type": "http.disconnect"}
 
             try:
@@ -140,7 +149,9 @@ async def process_messages() -> None:
                 async with asgi_receive_channel:
                     async for message in asgi_receive_channel:
                         if message["type"] == "http.response.start":
-                            assert not response_started
+                            if response_started:
+                                # Ignore duplicate response.start from ASGI app during SSE disconnect
+                                continue
                             status_code = message["status"]
                             response_headers = message.get("headers", [])
                             response_started = True
@@ -176,7 +187,7 @@ async def process_messages() -> None:
         return Response(
             status_code,
             headers=response_headers,
-            stream=StreamingASGIResponseStream(content_receive_channel),
+            stream=StreamingASGIResponseStream(content_receive_channel, send_disconnect),
         )
 
 
@@ -192,12 +203,18 @@ class StreamingASGIResponseStream(AsyncByteStream):
     def __init__(
         self,
         receive_channel: anyio.streams.memory.MemoryObjectReceiveStream[bytes],
+        send_disconnect: Callable[[], Awaitable[None]],
     ) -> None:
         self.receive_channel = receive_channel
+        self.send_disconnect = send_disconnect
 
     async def __aiter__(self) -> typing.AsyncIterator[bytes]:
         try:
             async for chunk in self.receive_channel:
                 yield chunk
         finally:
-            await self.receive_channel.aclose()
+            await self.aclose()
+
+    async def aclose(self) -> None:
+        await self.receive_channel.aclose()
+        await self.send_disconnect()

tests/conftest.py

@@ -1,6 +1,45 @@
+import anyio
 import pytest
+import sse_starlette
+from packaging import version
 
 
 @pytest.fixture
 def anyio_backend():
     return "asyncio"
+
+
+SSE_STARLETTE_VERSION = version.parse(sse_starlette.__version__)
+NEEDS_RESET = SSE_STARLETTE_VERSION < version.parse("3.0.0")
+
+
+@pytest.fixture(autouse=True)
+def reset_sse_app_status():
+    """Reset sse-starlette's global AppStatus singleton before each test.
+
+    AppStatus.should_exit_event is a global asyncio.Event that gets bound to
+    an event loop. This ensures each test gets a fresh Event and prevents
+    RuntimeError("bound to a different event loop") during parallel test
+    execution with pytest-xdist.
+
+    NOTE: This fixture is only necessary for sse-starlette < 3.0.0.
+    Version 3.0+ eliminated the global state issue entirely by using
+    context-local events instead of module-level singletons, providing
+    automatic test isolation without manual cleanup.
+
+    See <https://github.com/sysid/sse-starlette/pull/141> for more details.
+    """
+    if not NEEDS_RESET:
+        yield
+        return
+
+    # lazy import to avoid import errors
+    from sse_starlette.sse import AppStatus
+
+    # Setup: Reset before test
+    AppStatus.should_exit_event = anyio.Event()  # type: ignore[attr-defined]
+
+    yield
+
+    # Teardown: Reset after test to prevent contamination
+    AppStatus.should_exit_event = anyio.Event()  # type: ignore[attr-defined]