From 56d6bdcf65ae83a990dc07b13d9af518f48ca085 Mon Sep 17 00:00:00 2001 From: Tarun Krishna Reddy Date: Mon, 12 Oct 2026 21:40:13 -0700 Subject: [PATCH] Forward server stderr to errlog via pipe instead of inheriting its fd Fixes modelcontextprotocol/python-sdk#156: passing errlog (default sys.stderr) directly as the subprocess stderr target breaks in Jupyter, where sys.stderr has no fileno (UnsupportedOperation: fileno), swallowing all server diagnostics. The server's stderr is now always a pipe, and a reader task forwards it to errlog, so fileno-less streams work. Writes go through anyio.to_thread to keep blocking I/O off the loop. --- src/mcp/client/stdio.py | 37 ++++++- src/mcp/os/win32/utilities.py | 17 +-- tests/client/test_stdio.py | 164 ++++++++++++++++++++++++++--- tests/transports/stdio/conftest.py | 4 +- 4 files changed, 192 insertions(+), 30 deletions(-) diff --git a/src/mcp/client/stdio.py b/src/mcp/client/stdio.py index 6a3ad12111..8b4047fef2 100644 --- a/src/mcp/client/stdio.py +++ b/src/mcp/client/stdio.py @@ -10,6 +10,7 @@ import logging import os +import subprocess import sys from collections.abc import AsyncGenerator from contextlib import asynccontextmanager, suppress @@ -127,7 +128,6 @@ async def stdio_client( command=command, args=server.args, env=get_default_environment() | (server.env or {}), - errlog=errlog, cwd=server.cwd, ) @@ -182,6 +182,27 @@ async def stdin_writer() -> None: finally: writer_done.set() + async def stderr_reader() -> None: + # The server's stderr is a pipe, never errlog's fd: errlog may be a + # stream without a fileno (e.g. Jupyter's sys.stderr), which the + # subprocess spawn cannot consume (gh-156). Forward it here instead. + assert process.stderr, "Opened process is missing stderr" + + def forward(chunk: str) -> None: + # errlog is an arbitrary TextIO; keep the blocking write off the loop. + errlog.write(chunk) + errlog.flush() + + stderr = TextReceiveStream(process.stderr, encoding=server.encoding, errors=server.encoding_error_handler) + try: + async for chunk in stderr: + await anyio.to_thread.run_sync(forward, chunk) + except (anyio.ClosedResourceError, anyio.BrokenResourceError, ConnectionError): + # Teardown noise during shutdown, a real failure otherwise; stderr + # carries only diagnostics, so the session is unaffected either way. + if not shutting_down: + logger.exception("Reading from the MCP server's stderr failed mid-session") + async def shutdown() -> None: """Winds the transport down: stop traffic, flush, stop the server, release the streams.""" # Unblock the reader into its drain: a server stuck writing stdout cannot @@ -201,6 +222,7 @@ async def shutdown() -> None: async with anyio.create_task_group() as tg: tg.start_soon(stdout_reader) tg.start_soon(stdin_writer) + tg.start_soon(stderr_reader) try: yield read_stream, write_stream finally: @@ -252,7 +274,7 @@ async def _stop_server_process(process: ServerProcess) -> None: The escalation order is spec text; timeouts and tree-wide scope are SDK policy: https://modelcontextprotocol.io/specification/2025-11-25/basic/lifecycle#shutdown """ - assert process.stdin and process.stdout, "server process is spawned with pipes" + assert process.stdin and process.stdout and process.stderr, "server process is spawned with pipes" await _close_pipe(process.stdin) if not await _wait_for_process_exit(process, PROCESS_TERMINATION_TIMEOUT): @@ -265,6 +287,8 @@ async def _stop_server_process(process: ServerProcess) -> None: close_process_job(process) # A kill survivor can hold the stdout pipe open; poison the reader anyway. await _close_pipe(process.stdout) + # Same for stderr: the reader only ever sees EOF, never a close it made. + await _close_pipe(process.stderr) _close_subprocess_transport(process) @@ -330,20 +354,23 @@ async def _create_platform_compatible_process( command: str, args: list[str], env: dict[str, str] | None = None, - errlog: TextIO = sys.stderr, cwd: Path | str | None = None, ) -> ServerProcess: """Spawns the server in its own kill scope. A new session/process group on POSIX, a Job Object on Windows. + + The server's stderr is always a pipe: the caller's errlog stream may not + have a fileno (e.g. Jupyter's sys.stderr), so it cannot be handed to the + subprocess directly. `stdio_client` forwards the pipe to errlog instead. """ if sys.platform == "win32": # pragma: no cover - return await create_windows_process(command, args, env, errlog, cwd) + return await create_windows_process(command, args, env, cwd) else: # pragma: lax no cover return await anyio.open_process( [command, *args], env=env, - stderr=errlog, + stderr=subprocess.PIPE, cwd=cwd, start_new_session=True, ) diff --git a/src/mcp/os/win32/utilities.py b/src/mcp/os/win32/utilities.py index 321fda8a66..e3681bdaf9 100644 --- a/src/mcp/os/win32/utilities.py +++ b/src/mcp/os/win32/utilities.py @@ -7,7 +7,7 @@ import weakref from contextlib import suppress from pathlib import Path -from typing import BinaryIO, TextIO, TypeAlias, cast +from typing import BinaryIO, TypeAlias, cast import anyio from anyio.abc import Process @@ -92,9 +92,11 @@ def __init__(self, popen_obj: subprocess.Popen[bytes]) -> None: self.popen: subprocess.Popen[bytes] = popen_obj stdin = popen_obj.stdin stdout = popen_obj.stdout + stderr = popen_obj.stderr self.stdin = FileWriteStream(cast(BinaryIO, stdin)) if stdin else None self.stdout = FileReadStream(cast(BinaryIO, stdout)) if stdout else None + self.stderr = FileReadStream(cast(BinaryIO, stderr)) if stderr else None async def wait(self) -> int: """Waits for exit by polling the Popen. @@ -137,7 +139,6 @@ async def create_windows_process( command: str, args: list[str], env: dict[str, str] | None = None, - errlog: TextIO | None = sys.stderr, cwd: Path | str | None = None, ) -> Process | FallbackProcess: """Creates a subprocess with Job Object support for tree termination. @@ -149,8 +150,11 @@ async def create_windows_process( with it; children spawned before the assignment completes are not captured (see the inline note below). + The server's stderr is always a pipe (see `_create_platform_compatible_process`): + the caller's errlog stream is forwarded it by `stdio_client` instead. + Returns: - Process | FallbackProcess: The spawned process with async stdin/stdout streams. + Process | FallbackProcess: The spawned process with async stdin/stdout/stderr streams. """ try: process = await anyio.open_process( @@ -158,12 +162,12 @@ async def create_windows_process( env=env, # Ensure we don't create console windows for each process creationflags=getattr(subprocess, "CREATE_NO_WINDOW", 0), - stderr=errlog, + stderr=subprocess.PIPE, cwd=cwd, ) except NotImplementedError: # Windows event loops without async subprocess support (SelectorEventLoop) - process = await _create_windows_fallback_process(command, args, env, errlog, cwd) + process = await _create_windows_fallback_process(command, args, env, cwd) # Children spawned before the assignment completes land outside the job # (membership is inherited at CreateProcess, never acquired retroactively); @@ -177,7 +181,6 @@ async def _create_windows_fallback_process( command: str, args: list[str], env: dict[str, str] | None = None, - errlog: TextIO | None = sys.stderr, cwd: Path | str | None = None, ) -> FallbackProcess: """Spawns via subprocess.Popen and wraps it in FallbackProcess.""" @@ -185,7 +188,7 @@ async def _create_windows_fallback_process( [command, *args], stdin=subprocess.PIPE, stdout=subprocess.PIPE, - stderr=errlog, + stderr=subprocess.PIPE, env=env, cwd=cwd, bufsize=0, # Unbuffered output diff --git a/tests/client/test_stdio.py b/tests/client/test_stdio.py index b64eb01902..94df5732bc 100644 --- a/tests/client/test_stdio.py +++ b/tests/client/test_stdio.py @@ -9,6 +9,7 @@ import errno import gc +import io import logging import math import os @@ -19,7 +20,7 @@ from contextlib import AsyncExitStack, suppress from pathlib import Path from types import SimpleNamespace -from typing import TextIO, cast +from typing import cast import anyio import anyio.abc @@ -94,8 +95,8 @@ async def aclose(self) -> None: raise self._process.stdin_aclose_error -class _FakeStdout: - """The fake process's stdout: delegates to the in-memory stream. +class _FakePipe: + """The fake process's stdout/stderr: delegates to the in-memory stream. Optionally surfaces the abrupt-death or close-time errors a real pipe can. """ @@ -135,8 +136,9 @@ async def aclose(self) -> None: class FakeProcess: """In-memory stand-in for the spawned server process. - `feed`/`close_stdout` drive its stdout, `written` records client writes, `exit` - and the error knobs replay death and pipe failure modes. + `feed`/`close_stdout` drive its stdout, `feed_stderr`/`close_stderr` drive + its stderr, `written` records client writes, `exit` and the error knobs + replay death and pipe failure modes. """ def __init__( @@ -149,14 +151,21 @@ def __init__( stdout_eof_error: Exception | None = None, stdout_aclose_error: Exception | None = None, on_stdout_receive: Callable[[], None] | None = None, + stderr_eof_error: Exception | None = None, ) -> None: self._stdout_send, stdout_receive = anyio.create_memory_object_stream[bytes](math.inf) - self.stdout = _FakeStdout( + self.stdout = _FakePipe( stdout_receive, eof_error=stdout_eof_error, aclose_error=stdout_aclose_error, on_receive=self._dispatch_stdout_receive, ) + self._stderr_send, stderr_receive = anyio.create_memory_object_stream[bytes](math.inf) + self.stderr = _FakePipe( + stderr_receive, + eof_error=stderr_eof_error, + on_receive=self._dispatch_stderr_receive, + ) self.pid = 424242 self.written: list[bytes] = [] self.stdin_closed = anyio.Event() @@ -167,6 +176,7 @@ def __init__( self.stdin_send_blocks = stdin_send_blocks self.stdin_send_gate = stdin_send_gate self.on_stdout_receive = on_stdout_receive + self.on_stderr_receive: Callable[[], None] | None = None self.stdin = _FakeStdin(self) def _dispatch_stdout_receive(self) -> None: @@ -174,6 +184,11 @@ def _dispatch_stdout_receive(self) -> None: if self.on_stdout_receive is not None: self.on_stdout_receive() + def _dispatch_stderr_receive(self) -> None: + # Late-bound so a test can assign `on_stderr_receive` after construction. + if self.on_stderr_receive is not None: + self.on_stderr_receive() + async def feed(self, data: bytes) -> None: """Make `data` readable on the fake process's stdout.""" await self._stdout_send.send(data) @@ -182,10 +197,19 @@ def close_stdout(self) -> None: """End the fake process's stdout, as the kernel does when it dies.""" self._stdout_send.close() + async def feed_stderr(self, data: bytes) -> None: + """Make `data` readable on the fake process's stderr.""" + await self._stderr_send.send(data) + + def close_stderr(self) -> None: + """End the fake process's stderr, as the kernel does when it dies.""" + self._stderr_send.close() + def exit(self, code: int = 0) -> None: - """Die: set the exit code and EOF stdout, as the kernel does.""" + """Die: set the exit code and EOF stdout/stderr, as the kernel does.""" self.returncode = code self.close_stdout() + self.close_stderr() def pending_stdout_chunks(self) -> int: """How many fed chunks the client has not yet pulled off the fake stdout.""" @@ -210,7 +234,6 @@ async def fake_spawn( command: str, args: list[str], env: dict[str, str] | None = None, - errlog: TextIO = sys.stderr, cwd: Path | str | None = None, ) -> FakeProcess: return process @@ -546,7 +569,6 @@ async def failing_spawn( command: str, args: list[str], env: dict[str, str] | None = None, - errlog: TextIO = sys.stderr, cwd: Path | str | None = None, ) -> FakeProcess: raise OSError(errno.EACCES, "Permission denied") @@ -631,7 +653,6 @@ async def hanging_spawn( command: str, args: list[str], env: dict[str, str] | None = None, - errlog: TextIO = sys.stderr, cwd: Path | str | None = None, ) -> FakeProcess: spawn_started.set() @@ -673,7 +694,6 @@ async def failing_spawn( command: str, args: list[str], env: dict[str, str] | None = None, - errlog: TextIO = sys.stderr, cwd: Path | str | None = None, ) -> FakeProcess: raise ValueError("embedded null byte") @@ -993,6 +1013,120 @@ async def test_a_failing_stdout_close_still_closes_the_transport_streams(monkeyp gc.collect() +# --------------------------------------------------------------------------- +# Server stderr forwarding (gh-156) +# --------------------------------------------------------------------------- + + +class _JupyterLikeStderr(io.StringIO): + """Mimics Jupyter's sys.stderr: write() works, fileno() raises. + + Signals each write so tests can await forwarding deterministically. Writes + arrive on a worker thread (the SDK forwards stderr via `to_thread`), so the + signal is a `threading.Event`. + """ + + def __init__(self) -> None: + super().__init__() + self._written = threading.Event() + + def write(self, s: str) -> int: + result = super().write(s) + self._written.set() + return result + + def fileno(self) -> int: + raise io.UnsupportedOperation("fileno") + + async def wait_written(self) -> None: + await anyio.to_thread.run_sync(self._written.wait) + + +@pytest.mark.anyio +async def test_server_stderr_is_forwarded_to_errlog(monkeypatch: pytest.MonkeyPatch) -> None: + """The server's stderr pipe is forwarded to errlog by a reader task. + + errlog is never handed to the subprocess as its stderr fd (which is what + broke Jupyter in gh-156); the reader task forwards the piped output instead. + """ + errlog = _JupyterLikeStderr() + process = FakeProcess(on_stdin_close=lambda: process.exit(0)) + install_fake_process(monkeypatch, process) + + with anyio.fail_after(5): + async with stdio_client(FAKE_PARAMS, errlog=errlog): + await process.feed_stderr(b"server says hi\n") + await errlog.wait_written() + + assert errlog.getvalue() == "server says hi\n" + + +@pytest.mark.anyio +async def test_a_mid_session_stderr_failure_is_logged_and_leaves_the_session_usable( + monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture +) -> None: + """A mid-session stderr read failure is logged but does not break the session. + + Unlike stdout, stderr carries only diagnostics: its loss ends the reader + task, not the session. + """ + ping = JSONRPCRequest(jsonrpc="2.0", id=1, method="ping") + process = FakeProcess( + on_stdin_close=lambda: process.exit(0), + stderr_eof_error=ConnectionResetError("stderr pipe failed mid-session"), + ) + install_fake_process(monkeypatch, process) + + with anyio.fail_after(5): + async with stdio_client(FAKE_PARAMS) as (read_stream, _): + process.close_stderr() # EOF the pipe mid-session: the error surfaces now + await anyio.wait_all_tasks_blocked() # let the stderr reader hit it + # The session is unaffected: stdout traffic still flows. + await process.feed(_line(ping)) + assert await _next_message(read_stream) == ping + + assert "stderr failed mid-session" in caplog.text + + +@pytest.mark.anyio +async def test_a_kill_racing_a_pending_stderr_read_is_swallowed_during_shutdown( + monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture +) -> None: + """A hard kill during a pending stderr read must not escape the context manager. + + The read surfaces ConnectionResetError on the proactor backend; being expected + teardown noise, it is not logged as an error either. + """ + process = FakeProcess(stderr_eof_error=ConnectionResetError("read torn down by kill")) + terminated = install_fake_process(monkeypatch, process) + + with anyio.fail_after(5): + async with stdio_client(FAKE_PARAMS): + pass # the fake ignores stdin closure, so shutdown must escalate + + assert terminated == [process] + assert not [record for record in caplog.records if record.levelno >= logging.ERROR] + + +@pytest.mark.anyio +async def test_a_fileno_less_errlog_still_receives_server_stderr() -> None: + """End-to-end gh-156: an errlog without a fileno (Jupyter's sys.stderr) works. + + The spawn must not hand errlog to the subprocess (which raised + UnsupportedOperation: fileno), and the server's stderr must still reach it. + """ + errlog = _JupyterLikeStderr() + server_params = StdioServerParameters( + command=sys.executable, + args=["-c", "import sys; sys.stderr.write('diagnostic\\n'); sys.stderr.flush()"], + ) + with anyio.fail_after(10): + async with stdio_client(server_params, errlog=errlog): + await errlog.wait_written() + + assert errlog.getvalue() == "diagnostic\n" + + @pytest.mark.anyio async def test_a_process_surviving_the_kill_escalation_is_logged_and_abandoned( monkeypatch: pytest.MonkeyPatch, @@ -1022,9 +1156,10 @@ async def stubborn_terminate(proc: FakeProcess) -> None: assert stubborn == [process] assert process.returncode is None assert "still alive after the kill escalation" in caplog.text - # The fake "survived", so nothing ever EOF'd its stdout pipe; release it here - # or its GC-time ResourceWarning would fail a later test. + # The fake "survived", so nothing ever EOF'd its pipes; release them here + # or their GC-time ResourceWarnings would fail a later test. process.close_stdout() + process.close_stderr() # --------------------------------------------------------------------------- @@ -1251,10 +1386,9 @@ async def recording_spawn( command: str, args: list[str], env: dict[str, str] | None = None, - errlog: TextIO = sys.stderr, cwd: Path | str | None = None, ) -> anyio.abc.Process | FallbackProcess: - process = await _create_platform_compatible_process(command, args, env, errlog, cwd) + process = await _create_platform_compatible_process(command, args, env, cwd) spawned.append(process) return process diff --git a/tests/transports/stdio/conftest.py b/tests/transports/stdio/conftest.py index e9601ac6d5..23bde2cb2b 100644 --- a/tests/transports/stdio/conftest.py +++ b/tests/transports/stdio/conftest.py @@ -11,7 +11,6 @@ from collections.abc import Generator from contextlib import suppress from pathlib import Path -from typing import TextIO import anyio.abc import pytest @@ -37,10 +36,9 @@ async def recording_spawn( command: str, args: list[str], env: dict[str, str] | None = None, - errlog: TextIO = sys.stderr, cwd: Path | str | None = None, ) -> anyio.abc.Process | FallbackProcess: - process = await _create_platform_compatible_process(command, args, env, errlog, cwd) + process = await _create_platform_compatible_process(command, args, env, cwd) spawned.append(process) return process