diff --git a/src/mcp/server/stdio.py b/src/mcp/server/stdio.py index 5c1459dff..b66c0a522 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 677a99356..2ab1e226f 100644 --- a/tests/server/test_stdio.py +++ b/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