fix: prevent CRLF translation in stdio_server on Windows

src/mcp/server/stdio.py

@@ -39,9 +39,9 @@ async def stdio_server(stdin: anyio.AsyncFile[str] | None = None, stdout: anyio.
     # python is platform-dependent (Windows is particularly problematic), so we
     # re-wrap the underlying binary stream to ensure UTF-8.
     if not stdin:
-        stdin = anyio.wrap_file(TextIOWrapper(sys.stdin.buffer, encoding="utf-8", errors="replace"))
+        stdin = anyio.wrap_file(TextIOWrapper(sys.stdin.buffer, encoding="utf-8", errors="replace", newline=""))
     if not stdout:
-        stdout = anyio.wrap_file(TextIOWrapper(sys.stdout.buffer, encoding="utf-8"))
+        stdout = anyio.wrap_file(TextIOWrapper(sys.stdout.buffer, encoding="utf-8", newline=""))
 
     read_stream_writer, read_stream = create_context_streams[SessionMessage | Exception](0)
     write_stream, write_stream_reader = create_context_streams[SessionMessage](0)

tests/server/test_stdio.py

@@ -1,10 +1,12 @@
 import io
 import sys
 from io import TextIOWrapper
+from typing import Any
 
 import anyio
 import pytest
 
+import mcp.server.stdio as stdio_module
 from mcp.server.stdio import stdio_server
 from mcp.shared.message import SessionMessage
 from mcp.types import JSONRPCMessage, JSONRPCRequest, JSONRPCResponse, jsonrpc_message_adapter
@@ -92,3 +94,29 @@ async def test_stdio_server_invalid_utf8(monkeypatch: pytest.MonkeyPatch):
                 second = await read_stream.receive()
                 assert isinstance(second, SessionMessage)
                 assert second.message == valid
+
+
+@pytest.mark.anyio
+async def test_stdio_server_disables_newline_translation(monkeypatch: pytest.MonkeyPatch):
+    raw_stdin = io.BytesIO()
+    raw_stdout = io.BytesIO()
+
+    monkeypatch.setattr(sys, "stdin", TextIOWrapper(raw_stdin, encoding="utf-8"))
+    monkeypatch.setattr(sys, "stdout", TextIOWrapper(raw_stdout, encoding="utf-8"))
+
+    calls: list[dict[str, str | None]] = []
+    real_text_io_wrapper = TextIOWrapper
+
+    def spy(buffer: Any, *args: Any, **kwargs: Any) -> TextIOWrapper:
+        calls.append({"errors": kwargs.get("errors"), "newline": kwargs.get("newline")})
+        return real_text_io_wrapper(buffer, *args, **kwargs)
+
+    monkeypatch.setattr(stdio_module, "TextIOWrapper", spy)
+
+    with anyio.fail_after(5):
+        async with stdio_server() as (read_stream, write_stream):
+            await write_stream.aclose()
+            await read_stream.aclose()
+
+    assert {"errors": "replace", "newline": ""} in calls
+    assert {"errors": None, "newline": ""} in calls