fix: JSONRPCDispatcher.run enters both streams; ClosedResourceError is EOF

run() now enters write_stream alongside read_stream so the write end is
released when the read loop exits (BaseSession did this; without it every
[sse] interaction leg leaks a MemoryObjectSendStream and fails under
filterwarnings=error).

ClosedResourceError from the read iterator is caught and treated as clean
EOF. Stateless SHTTP teardown closes the dispatcher's receive end after
the request is handled; the next __anext__ call on the now-closed stream
raises, which previously surfaced as 'Stateless session crashed'.

Author: maxisbey

src/mcp/shared/jsonrpc_dispatcher.py

@@ -337,13 +337,19 @@ async def run(
                 self._tg = tg
                 self._running = True
                 task_status.started()
-                async with self._read_stream:
-                    async for item in self._read_stream:
-                        # Duck-typed: `_context_streams.ContextReceiveStream`
-                        # exposes `.last_context` (the sender's contextvars
-                        # snapshot per message). Plain memory streams don't.
-                        sender_ctx: contextvars.Context | None = getattr(self._read_stream, "last_context", None)
-                        self._dispatch(item, on_request, on_notify, sender_ctx)
+                async with self._read_stream, self._write_stream:
+                    try:
+                        async for item in self._read_stream:
+                            # Duck-typed: `_context_streams.ContextReceiveStream`
+                            # exposes `.last_context` (the sender's contextvars
+                            # snapshot per message). Plain memory streams don't.
+                            sender_ctx: contextvars.Context | None = getattr(self._read_stream, "last_context", None)
+                            self._dispatch(item, on_request, on_notify, sender_ctx)
+                    except anyio.ClosedResourceError:
+                        # The transport closed our receive end and we looped back
+                        # to `__anext__` on the now-closed stream (stateless SHTTP
+                        # teardown). Same as EOF.
+                        logger.debug("read stream closed by transport; treating as EOF")
                 # Read stream EOF: wake any blocked `send_raw_request` waiters now,
                 # *before* the task group joins, so handlers parked in
                 # `dctx.send_raw_request()` can unwind and the join doesn't deadlock.

tests/shared/test_jsonrpc_dispatcher.py

@@ -176,6 +176,43 @@ async def caller() -> None:
             s.close()
 
 
+@pytest.mark.anyio
+async def test_run_returns_cleanly_when_read_stream_receive_end_is_closed():
+    """Iterating a closed receive end raises ClosedResourceError; run() treats it as EOF.
+
+    Stateless SHTTP teardown closes the dispatcher's receive end after the
+    request is handled; the next loop iteration must not surface as a crash.
+    """
+    c2s_send, c2s_recv = anyio.create_memory_object_stream[SessionMessage | Exception](32)
+    s2c_send, s2c_recv = anyio.create_memory_object_stream[SessionMessage | Exception](32)
+    server: JSONRPCDispatcher[TransportContext] = JSONRPCDispatcher(c2s_recv, s2c_send)
+    on_request, on_notify = echo_handlers(Recorder())
+    # Close the dispatcher's own receive end (not the send end) before run()
+    # iterates it: __anext__ on a closed stream raises ClosedResourceError.
+    c2s_recv.close()
+    with anyio.fail_after(5):
+        await server.run(on_request, on_notify)
+    for s in (c2s_send, s2c_send, s2c_recv):
+        s.close()
+
+
+@pytest.mark.anyio
+async def test_run_closes_write_stream_on_exit():
+    """run() enters both streams; the write end is released on EOF."""
+    c2s_send, c2s_recv = anyio.create_memory_object_stream[SessionMessage | Exception](32)
+    s2c_send, s2c_recv = anyio.create_memory_object_stream[SessionMessage | Exception](32)
+    server: JSONRPCDispatcher[TransportContext] = JSONRPCDispatcher(c2s_recv, s2c_send)
+    on_request, on_notify = echo_handlers(Recorder())
+    async with anyio.create_task_group() as tg:
+        await tg.start(server.run, on_request, on_notify)
+        c2s_send.close()  # EOF the read side; run() exits
+        with anyio.fail_after(5):
+            # Write end was entered and released by run(); peer's receive sees EOF.
+            with pytest.raises(anyio.EndOfStream):
+                await s2c_recv.receive()
+    s2c_recv.close()
+
+
 @pytest.mark.anyio
 async def test_late_response_after_timeout_is_dropped_without_crashing():
     handler_started = anyio.Event()