fix: Handle BrokenResourceError on Windows Python 3.13

- Improve stream cleanup order in Windows stdio client
- Add comprehensive exception handling for resource errors
- Handle process termination edge cases better
- Prevent race conditions in async stream cleanup

Author: spacelord16

src/mcp/client/stdio/__init__.py

@@ -115,7 +115,11 @@ async def stdio_client(server: StdioServerParameters, errlog: TextIO = sys.stder
         process = await _create_platform_compatible_process(
             command=command,
             args=server.args,
-            env=({**get_default_environment(), **server.env} if server.env is not None else get_default_environment()),
+            env=(
+                {**get_default_environment(), **server.env}
+                if server.env is not None
+                else get_default_environment()
+            ),
             errlog=errlog,
             cwd=server.cwd,
         )
@@ -150,7 +154,7 @@ async def stdout_reader():
 
                         session_message = SessionMessage(message)
                         await read_stream_writer.send(session_message)
-        except anyio.ClosedResourceError:
+        except (anyio.ClosedResourceError, anyio.BrokenResourceError):
             await anyio.lowlevel.checkpoint()
 
     async def stdin_writer():
@@ -159,14 +163,16 @@ async def stdin_writer():
         try:
             async with write_stream_reader:
                 async for session_message in write_stream_reader:
-                    json = session_message.message.model_dump_json(by_alias=True, exclude_none=True)
+                    json = session_message.message.model_dump_json(
+                        by_alias=True, exclude_none=True
+                    )
                     await process.stdin.send(
                         (json + "\n").encode(
                             encoding=server.encoding,
                             errors=server.encoding_error_handler,
                         )
                     )
-        except anyio.ClosedResourceError:
+        except (anyio.ClosedResourceError, anyio.BrokenResourceError):
             await anyio.lowlevel.checkpoint()
 
     async with (
@@ -184,13 +190,30 @@ async def stdin_writer():
                     await terminate_windows_process(process)
                 else:
                     process.terminate()
-            except ProcessLookupError:
-                # Process already exited, which is fine
+            except (ProcessLookupError, OSError, anyio.BrokenResourceError):
+                # Process already exited or couldn't be terminated, which is fine
+                pass
+
+            # Close streams in proper order to avoid BrokenResourceError
+            try:
+                await read_stream.aclose()
+            except (anyio.ClosedResourceError, anyio.BrokenResourceError):
+                pass
+
+            try:
+                await write_stream.aclose()
+            except (anyio.ClosedResourceError, anyio.BrokenResourceError):
+                pass
+
+            try:
+                await read_stream_writer.aclose()
+            except (anyio.ClosedResourceError, anyio.BrokenResourceError):
+                pass
+
+            try:
+                await write_stream_reader.aclose()
+            except (anyio.ClosedResourceError, anyio.BrokenResourceError):
                 pass
-            await read_stream.aclose()
-            await write_stream.aclose()
-            await read_stream_writer.aclose()
-            await write_stream_reader.aclose()
 
 
 def _get_executable_command(command: str) -> str:
@@ -223,6 +246,8 @@ async def _create_platform_compatible_process(
     if sys.platform == "win32":
         process = await create_windows_process(command, args, env, errlog, cwd)
     else:
-        process = await anyio.open_process([command, *args], env=env, stderr=errlog, cwd=cwd)
+        process = await anyio.open_process(
+            [command, *args], env=env, stderr=errlog, cwd=cwd
+        )
 
     return process

src/mcp/client/stdio/win32.py

@@ -62,8 +62,12 @@ def __init__(self, popen_obj: subprocess.Popen[bytes]):
         self.stdout_raw = popen_obj.stdout  # type: ignore[assignment]
         self.stderr = popen_obj.stderr  # type: ignore[assignment]
 
-        self.stdin = FileWriteStream(cast(BinaryIO, self.stdin_raw)) if self.stdin_raw else None
-        self.stdout = FileReadStream(cast(BinaryIO, self.stdout_raw)) if self.stdout_raw else None
+        self.stdin = (
+            FileWriteStream(cast(BinaryIO, self.stdin_raw)) if self.stdin_raw else None
+        )
+        self.stdout = (
+            FileReadStream(cast(BinaryIO, self.stdout_raw)) if self.stdout_raw else None
+        )
 
     async def __aenter__(self):
         """Support async context manager entry."""
@@ -76,20 +80,50 @@ async def __aexit__(
         exc_tb: object | None,
     ) -> None:
         """Terminate and wait on process exit inside a thread."""
-        self.popen.terminate()
-        await to_thread.run_sync(self.popen.wait)
+        try:
+            self.popen.terminate()
+            await to_thread.run_sync(self.popen.wait)
+        except (ProcessLookupError, OSError):
+            # Process already exited or couldn't be terminated, which is fine
+            pass
 
         # Close the file handles to prevent ResourceWarning
-        if self.stdin:
-            await self.stdin.aclose()
-        if self.stdout:
-            await self.stdout.aclose()
-        if self.stdin_raw:
-            self.stdin_raw.close()
-        if self.stdout_raw:
-            self.stdout_raw.close()
-        if self.stderr:
-            self.stderr.close()
+        # Close in reverse order of creation to avoid BrokenResourceError
+        try:
+            if self.stderr:
+                self.stderr.close()
+        except (OSError, ValueError):
+            # Stream already closed or invalid, ignore
+            pass
+
+        try:
+            if self.stdout_raw:
+                self.stdout_raw.close()
+        except (OSError, ValueError):
+            # Stream already closed or invalid, ignore
+            pass
+
+        try:
+            if self.stdin_raw:
+                self.stdin_raw.close()
+        except (OSError, ValueError):
+            # Stream already closed or invalid, ignore
+            pass
+
+        # Close async stream wrappers
+        try:
+            if self.stdout:
+                await self.stdout.aclose()
+        except (anyio.ClosedResourceError, anyio.BrokenResourceError):
+            # Stream already closed, ignore
+            pass
+
+        try:
+            if self.stdin:
+                await self.stdin.aclose()
+        except (anyio.ClosedResourceError, anyio.BrokenResourceError):
+            # Stream already closed, ignore
+            pass
 
     async def wait(self):
         """Async wait for process completion."""
@@ -175,8 +209,19 @@ async def terminate_windows_process(process: Process | FallbackProcess):
     """
     try:
         process.terminate()
-        with anyio.fail_after(2.0):
-            await process.wait()
-    except TimeoutError:
-        # Force kill if it doesn't terminate
-        process.kill()
+        try:
+            with anyio.fail_after(2.0):
+                await process.wait()
+        except TimeoutError:
+            # Force kill if it doesn't terminate
+            try:
+                process.kill()
+                # Give it a moment to actually terminate after kill
+                with anyio.fail_after(1.0):
+                    await process.wait()
+            except (TimeoutError, ProcessLookupError, OSError):
+                # Process is really stubborn or already gone, just continue
+                pass
+    except (ProcessLookupError, OSError, anyio.BrokenResourceError):
+        # Process already exited or couldn't be terminated, which is fine
+        pass