modelcontextprotocol
diff --git a/src/mcp/client/stdio.py b/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.
 
@@ -151,7 +151,7 @@ async def stdout_reader():
                     for line in lines:
                         try:
                             message = types.jsonrpc_message_adapter.validate_json(line, by_name=False)
-                        except Exception as exc:  # pragma: no cover
+                        except Exception as exc:
                             logger.exception("Failed to parse JSONRPC message from server")
                             await read_stream_writer.send(exc)
                             continue

diff --git a/tests/client/test_stdio.py b/tests/client/test_stdio.py
@@ -501,6 +501,40 @@ async def test_stdio_client_graceful_stdin_exit():
     )
 
 
+@pytest.mark.anyio
+async def test_stdio_client_invalid_utf8():
+    """Malformed UTF-8 from the child must not crash the transport.
+
+    Invalid bytes are replaced with U+FFFD (default encoding_error_handler),
+    which fails JSON parsing and arrives as an in-stream exception.  The
+    following valid JSON-RPC line must still be delivered as a SessionMessage.
+    """
+    valid_line = '{"jsonrpc":"2.0","id":1,"method":"ping"}'
+    script = textwrap.dedent(
+        f"""
+        import sys
+        sys.stdout.buffer.write(b"\\xff\\xfe\\n")
+        sys.stdout.buffer.write({valid_line!r}.encode() + b"\\n")
+        sys.stdout.buffer.flush()
+        """
+    )
+
+    server_params = StdioServerParameters(
+        command=sys.executable,
+        args=["-c", script],
+    )
+
+    items: list[SessionMessage | Exception] = []
+
+    with anyio.fail_after(5.0):
+        async with stdio_client(server_params) as (read_stream, _write_stream):
+            items.extend([item async for item in read_stream])
+
+    assert len(items) == 2
+    assert isinstance(items[0], Exception)
+    assert isinstance(items[1], SessionMessage)
+
+
 @pytest.mark.anyio
 async def test_stdio_client_stdin_close_ignored():
     """Test that when a process ignores stdin closure, the shutdown sequence