From 432746b8649b925b8322585b038ac27b81ebfda2 Mon Sep 17 00:00:00 2001
From: Jianke LIN <crayozakka@gmail.com>
Date: Sun, 31 May 2026 22:10:41 +0200
Subject: [PATCH 1/2] fix: prevent CRLF in stdio_server

---
 src/mcp/server/stdio.py    |  4 ++--
 tests/server/test_stdio.py | 27 +++++++++++++++++++++++++++
 2 files changed, 29 insertions(+), 2 deletions(-)

diff --git a/src/mcp/server/stdio.py b/src/mcp/server/stdio.py
index 5c1459dff6..b66c0a5224 100644
--- a/src/mcp/server/stdio.py
+++ b/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)
diff --git a/tests/server/test_stdio.py b/tests/server/test_stdio.py
index 677a993567..4af63fe6aa 100644
--- a/tests/server/test_stdio.py
+++ b/tests/server/test_stdio.py
@@ -5,6 +5,7 @@
 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 +93,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, object | None]] = []
+    real_text_io_wrapper = TextIOWrapper
+
+    def spy(buffer: io.BufferedIOBase, *args: object, **kwargs: object) -> 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

From d6375cd6980758198dd7f9e017af1d8310739061 Mon Sep 17 00:00:00 2001
From: Jianke LIN <crayozakka@gmail.com>
Date: Sun, 31 May 2026 22:17:03 +0200
Subject: [PATCH 2/2] test: make stdio newline spy pyright-friendly

---
 tests/server/test_stdio.py | 5 +++--
 1 file changed, 3 insertions(+), 2 deletions(-)

diff --git a/tests/server/test_stdio.py b/tests/server/test_stdio.py
index 4af63fe6aa..2ab1e226f9 100644
--- a/tests/server/test_stdio.py
+++ b/tests/server/test_stdio.py
@@ -1,6 +1,7 @@
 import io
 import sys
 from io import TextIOWrapper
+from typing import Any
 
 import anyio
 import pytest
@@ -103,10 +104,10 @@ async def test_stdio_server_disables_newline_translation(monkeypatch: pytest.Mon
     monkeypatch.setattr(sys, "stdin", TextIOWrapper(raw_stdin, encoding="utf-8"))
     monkeypatch.setattr(sys, "stdout", TextIOWrapper(raw_stdout, encoding="utf-8"))
 
-    calls: list[dict[str, object | None]] = []
+    calls: list[dict[str, str | None]] = []
     real_text_io_wrapper = TextIOWrapper
 
-    def spy(buffer: io.BufferedIOBase, *args: object, **kwargs: object) -> 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)