Prevent malformed child stdout from crashing stdio_client

The stdio server path already degrades invalid UTF-8 into parse errors instead of killing the transport, but the client side still defaulted to strict decoding and could crash the whole task group on a single bad line from a child process. This changes the default decode strategy to replacement, broadens shutdown-time exception handling for abrupt child exits, and adds a regression test that proves malformed output becomes an in-stream error while the next valid JSON-RPC message still arrives.

Constraint: Must preserve stdio_client cleanup behavior when subprocesses exit early or reset pipes
Rejected: Add a new configuration flag only for the regression test | leaves the default transport behavior asymmetric and crash-prone
Confidence: high
Scope-risk: narrow
Reversibility: clean
Directive: Keep client and server stdio malformed-UTF-8 handling symmetric unless the protocol deliberately diverges
Tested: uv run --frozen pytest tests/client/test_stdio.py; uv run --frozen ruff check src/mcp/client/stdio.py tests/client/test_stdio.py; uv run --frozen pyright src/mcp/client/stdio.py
Not-tested: Full multi-platform matrix outside local macOS / Python 3.10 run

Author: shaun0927

src/mcp/client/stdio.py

@@ -92,7 +92,7 @@ class StdioServerParameters(BaseModel):
     Defaults to utf-8.
     """
 
-    encoding_error_handler: Literal["strict", "ignore", "replace"] = "strict"
+    encoding_error_handler: Literal["strict", "ignore", "replace"] = "replace"
     """
     The text encoding error handler.
 
@@ -158,7 +158,7 @@ async def stdout_reader():
 
                         session_message = SessionMessage(message)
                         await read_stream_writer.send(session_message)
-        except anyio.ClosedResourceError:  # pragma: lax no cover
+        except (anyio.ClosedResourceError, anyio.BrokenResourceError, ConnectionResetError):  # pragma: lax no cover
             await anyio.lowlevel.checkpoint()
 
     async def stdin_writer():
@@ -174,7 +174,7 @@ async def stdin_writer():
                             errors=server.encoding_error_handler,
                         )
                     )
-        except anyio.ClosedResourceError:  # pragma: no cover
+        except (anyio.ClosedResourceError, anyio.BrokenResourceError, ConnectionResetError):  # pragma: no cover
             await anyio.lowlevel.checkpoint()
 
     async with anyio.create_task_group() as tg, process:

tests/client/test_stdio.py

@@ -4,6 +4,7 @@
 import textwrap
 import time
 from contextlib import AsyncExitStack, suppress
+from pathlib import Path
 
 import anyio
 import anyio.abc
@@ -70,6 +71,42 @@ async def test_stdio_client():
         assert read_messages[1] == JSONRPCResponse(jsonrpc="2.0", id=2, result={})
 
 
+@pytest.mark.anyio
+async def test_stdio_client_invalid_utf8_from_server_does_not_crash(tmp_path: Path):
+    """A buggy child server should surface malformed UTF-8 as an in-stream error.
+
+    The client should continue reading subsequent valid JSON-RPC lines instead of
+    crashing the whole transport task group during decoding.
+    """
+    script = tmp_path / "bad_stdout_server.py"
+    valid = JSONRPCRequest(jsonrpc="2.0", id=1, method="ping")
+    script.write_text(
+        textwrap.dedent(
+            f"""\
+            import sys
+            import time
+
+            sys.stdout.buffer.write(b"\\xff\\xfe\\n")
+            sys.stdout.buffer.write({valid.model_dump_json(by_alias=True, exclude_none=True)!r}.encode() + b"\\n")
+            sys.stdout.buffer.flush()
+            time.sleep(0.2)
+            """
+        )
+    )
+
+    server_params = StdioServerParameters(command=sys.executable, args=[str(script)])
+
+    with anyio.fail_after(5):
+        async with stdio_client(server_params) as (read_stream, write_stream):
+            await write_stream.aclose()
+            first = await read_stream.receive()
+            assert isinstance(first, Exception)
+
+            second = await read_stream.receive()
+            assert isinstance(second, SessionMessage)
+            assert second.message == valid
+
+
 @pytest.mark.anyio
 async def test_stdio_client_bad_path():
     """Check that the connection doesn't hang if process errors."""