From 5ac79b46493b22332f9f0a4179ab0164e71d37e6 Mon Sep 17 00:00:00 2001 From: Jianke LIN 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 054a157b3b..c99f4deb1c 100644 --- a/tests/server/test_stdio.py +++ b/tests/server/test_stdio.py @@ -8,6 +8,7 @@ import anyio import pytest +import mcp.server.stdio as stdio_module from mcp.server.mcpserver import MCPServer from mcp.server.stdio import stdio_server from mcp.shared.message import SessionMessage @@ -96,6 +97,32 @@ async def test_stdio_server_invalid_utf8(monkeypatch: pytest.MonkeyPatch) -> Non 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 + + class _KeepOpenBytesIO(io.BytesIO): """A BytesIO that survives its TextIOWrapper being closed. From 0bbd04a4770dd7d556004508f01a17bb942a2fa3 Mon Sep 17 00:00:00 2001 From: Jianke LIN 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 c99f4deb1c..aa913dc36f 100644 --- a/tests/server/test_stdio.py +++ b/tests/server/test_stdio.py @@ -4,6 +4,7 @@ from collections.abc import AsyncIterator from contextlib import asynccontextmanager from io import TextIOWrapper +from typing import Any import anyio import pytest @@ -105,10 +106,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)