fix(stdio): serve bufferless std streams as text instead of crashing

_claim_fd falls back to `stream.buffer` whenever the stream is not backed
by the expected descriptor. But _is_backed_by_fd also reports False when
the stream has no `.buffer` at all, so that fallback dereferences an
attribute it just proved might be missing.

A sys.stdin/sys.stdout replaced with io.StringIO (test harnesses, and
embedded hosts that swap the std streams) therefore raised AttributeError
before serving a single message.

Return None for the buffer in that case and serve the text stream in
place: it is already text, so there is no binary layer to re-encode and
none for _UnownedTextWrapper to protect from close.

Signed-off-by: Mike German <mike@stepsventures.com>

Author: steps-re

src/mcp/server/stdio.py

@@ -103,19 +103,34 @@ def _restore_fd(fd: int, private_fd: int) -> bool:
     return True
 
 
+def _text_transport(stream: TextIO, buffer: BinaryIO | None, errors: str | None = None) -> anyio.AsyncFile[str]:
+    """Serve the wire as UTF-8 text.
+
+    A stream with no buffer at all (io.StringIO under a test harness, or an
+    embedded host that replaced sys.stdout) is already text and owns no binary
+    layer to re-encode or to protect from close, so it is served in place.
+    """
+    if buffer is None:
+        return anyio.wrap_file(stream)
+    return anyio.wrap_file(_UnownedTextWrapper(buffer, encoding="utf-8", errors=errors))
+
+
 def _claim_fd(
     fd: int, stream: TextIO, mode: Literal["rb", "wb"], open_diversion: Callable[[], int]
-) -> tuple[BinaryIO, Callable[[], None] | None]:
+) -> tuple[BinaryIO | None, Callable[[], None] | None]:
     """Claim a standard stream: divert fd and serve the wire from a private duplicate.
 
     Best-effort: when descriptors cannot be duplicated or diverted, serves the
     sys stream's buffer in place, exactly as v1 did, with the claim held.
 
+    Returns a None buffer when the stream exposes no binary layer, which means the
+    caller must serve it as text; every other path returns a binary stream.
+
     Raises:
         RuntimeError: fd is already claimed by another transport in this process.
     """
     if not _is_backed_by_fd(stream, fd):
-        return stream.buffer, None
+        return getattr(stream, "buffer", None), None
     claim = _StreamClaim(fd)
     with _claims_lock:
         if fd in _claims:
@@ -173,10 +188,10 @@ async def stdio_server(stdin: anyio.AsyncFile[str] | None = None, stdout: anyio.
     try:
         if not stdin:
             stdin_buffer, restore_stdin = _claim_fd(0, sys.stdin, "rb", _open_stdin_diversion)
-            stdin = anyio.wrap_file(_UnownedTextWrapper(stdin_buffer, encoding="utf-8", errors="replace"))
+            stdin = _text_transport(sys.stdin, stdin_buffer, errors="replace")
         if not stdout:
             stdout_buffer, restore_stdout = _claim_fd(1, sys.stdout, "wb", _open_stdout_diversion)
-            stdout = anyio.wrap_file(_UnownedTextWrapper(stdout_buffer, encoding="utf-8"))
+            stdout = _text_transport(sys.stdout, stdout_buffer)
 
         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

@@ -97,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_serves_bufferless_std_streams_in_place(monkeypatch: pytest.MonkeyPatch) -> None:
+    """A sys.stdin/sys.stdout with no .buffer is served as text rather than crashing.
+
+    Test harnesses and embedded hosts routinely replace the std streams with
+    io.StringIO, which exposes no binary layer for the claim path to re-encode.
+    """
+    request = JSONRPCRequest(jsonrpc="2.0", id=1, method="ping")
+    stdout = io.StringIO()
+    monkeypatch.setattr(sys, "stdin", io.StringIO(request.model_dump_json(by_alias=True, exclude_none=True) + "\n"))
+    monkeypatch.setattr(sys, "stdout", stdout)
+
+    with anyio.fail_after(5):
+        async with stdio_server() as (read_stream, write_stream):
+            async with read_stream:  # pragma: no branch
+                received = await read_stream.receive()
+                assert isinstance(received, SessionMessage)
+                assert received.message == request
+
+            response = JSONRPCResponse(jsonrpc="2.0", id=1, result={})
+            async with write_stream:
+                await write_stream.send(SessionMessage(response))
+
+    assert jsonrpc_message_adapter.validate_json(stdout.getvalue(), by_name=False) == response
+
+
 @contextmanager
 def _pipe_planted_on_fd0(monkeypatch: pytest.MonkeyPatch) -> Iterator[tuple[int, int]]:
     """Plants a fresh pipe on fd 0 and rebinds sys.stdin over it; yields (read_fd, write_fd).