fix: handle BrokenResourceError in stdio_client

Author: Pavithra-Madhan

src/mcp/client/session.py

@@ -144,6 +144,11 @@ def __init__(
         self._task_handlers = experimental_task_handlers or ExperimentalTaskHandlers()
 
     async def initialize(self) -> types.InitializeResult:
+        if not getattr(self, "_entered", False):
+            raise RuntimeError(
+                "ClientSession must be used inside `async with` before calling initialize()"
+                )
+
         sampling = types.SamplingCapability() if self._sampling_callback is not _default_sampling_callback else None
         elicitation = (
             types.ElicitationCapability(

src/mcp/client/stdio/__init__.py

@@ -155,11 +155,17 @@ async def stdout_reader():
                             message = types.JSONRPCMessage.model_validate_json(line)
                         except Exception as exc:  # pragma: no cover
                             logger.exception("Failed to parse JSONRPC message from server")
-                            await read_stream_writer.send(exc)
+                            try:
+                                await read_stream_writer.send(exc)
+                            except (anyio.BrokenResourceError, anyio.ClosedResourceError):
+                                return
                             continue
 
                         session_message = SessionMessage(message)
-                        await read_stream_writer.send(session_message)
+                        try:
+                            await read_stream_writer.send(session_message)
+                        except (anyio.BrokenResourceError, anyio.ClosedResourceError):
+                            return
         except anyio.ClosedResourceError:  # pragma: no cover
             await anyio.lowlevel.checkpoint()
 

src/mcp/shared/session.py

@@ -219,6 +219,7 @@ def add_response_router(self, router: ResponseRouter) -> None:
         self._response_routers.append(router)
 
     async def __aenter__(self) -> Self:
+        self._entered = True
         self._task_group = anyio.create_task_group()
         await self._task_group.__aenter__()
         self._task_group.start_soon(self._receive_loop)