diff --git a/src/mcp/client/stdio.py b/src/mcp/client/stdio.py index 3e03eef9ef..304e71395b 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 @@ -68,6 +69,9 @@ # Time for the writer to flush accepted messages before stdin closes. _WRITER_FLUSH_TIMEOUT = 0.5 +# Time for the forwarder to drain the dead server's remaining stderr into errlog. +_STDERR_DRAIN_TIMEOUT = 0.5 + # How often to poll returncode while waiting for the process to die. _EXIT_POLL_INTERVAL = 0.01 @@ -122,11 +126,16 @@ async def stdio_client( """ command = _get_executable_command(server.command) + # A child process inherits stderr as an OS file descriptor. Writer objects that + # have none (Jupyter's ipykernel stream, io.StringIO) cannot be inherited, so the + # server's stderr is piped back and forwarded into errlog by a reader task instead. + forward_stderr = _lacks_file_descriptor(errlog) + process = await _create_platform_compatible_process( command=command, args=server.args, env=get_default_environment() | (server.env or {}), - errlog=errlog, + errlog=subprocess.PIPE if forward_stderr else errlog, cwd=server.cwd, ) @@ -137,6 +146,7 @@ async def stdio_client( shutting_down = False writer_done = anyio.Event() + stderr_done = anyio.Event() async def stdout_reader() -> None: assert process.stdout, "Opened process is missing stdout" @@ -165,6 +175,30 @@ async def stdout_reader() -> None: if not shutting_down: logger.exception("Reading from the MCP server's stdout failed mid-session") + async def stderr_reader() -> None: + """Forwards the server's piped stderr into errlog. + + Only runs when errlog could not be inherited by the child; keeps the + server's diagnostics visible where a bare fd hand-off would have lost them. + """ + assert process.stderr, "Piped stderr is missing from the opened process" + + stderr = TextReceiveStream(process.stderr, encoding=server.encoding, errors="replace") + try: + async for chunk in stderr: + errlog.write(chunk) + errlog.flush() + except (anyio.ClosedResourceError, anyio.BrokenResourceError, ConnectionError): + pass # the pipe went away with the process; nothing left to forward + except ValueError: + # errlog was closed under us (a notebook cell finishing, a StringIO + # released). The server's own traffic must not fail over a lost log sink. + logger.debug("Stopped forwarding the MCP server's stderr: the log stream was closed") + finally: + # Reaching EOF means the server's last diagnostics are in errlog, which + # is what shutdown waits on before cancelling this task. + stderr_done.set() + async def stdin_writer() -> None: assert process.stdin, "Opened process is missing stdin" @@ -193,6 +227,11 @@ async def shutdown() -> None: if flush_scope.cancelled_caught: await anyio.lowlevel.cancel_shielded_checkpoint() # resync coverage on 3.11 (gh-106749) await _stop_server_process(process) + # The server is dead, so its stderr pipe is at EOF with at most a buffer left; + # let the forwarder finish it before the task group's cancel takes the task out. + if forward_stderr: + with anyio.move_on_after(_STDERR_DRAIN_TIMEOUT): + await stderr_done.wait() await _aclose_all(read_stream, write_stream, read_stream_writer, write_stream_reader) # One pass so unblocked tasks exit via their except paths before the cancel. await anyio.lowlevel.checkpoint() @@ -200,6 +239,8 @@ async def shutdown() -> None: async with anyio.create_task_group() as tg: tg.start_soon(stdout_reader) tg.start_soon(stdin_writer) + if forward_stderr: + tg.start_soon(stderr_reader) try: yield read_stream, write_stream finally: @@ -317,6 +358,22 @@ def _close_subprocess_transport(process: ServerProcess) -> None: close() +def _lacks_file_descriptor(errlog: TextIO) -> bool: + """Reports whether errlog has no OS file descriptor for a child to inherit. + + Jupyter's ipykernel stream and io.StringIO answer fileno() with an error; + real files, pipes and terminals return one. ipykernel does expose a + descriptor once fd capture is on, and that path already reaches the + notebook, so inheriting it stays correct. + """ + try: + errlog.fileno() + except (AttributeError, OSError, ValueError): + # io.UnsupportedOperation derives from OSError and ValueError. + return True + return False + + def _get_executable_command(command: str) -> str: """Normalizes the command for the current platform.""" if sys.platform == "win32": # pragma: no cover @@ -329,7 +386,7 @@ async def _create_platform_compatible_process( command: str, args: list[str], env: dict[str, str] | None = None, - errlog: TextIO = sys.stderr, + errlog: TextIO | int = sys.stderr, cwd: Path | str | None = None, ) -> ServerProcess: """Spawns the server in its own kill scope. diff --git a/src/mcp/os/win32/utilities.py b/src/mcp/os/win32/utilities.py index 321fda8a66..bffe87abd4 100644 --- a/src/mcp/os/win32/utilities.py +++ b/src/mcp/os/win32/utilities.py @@ -92,9 +92,12 @@ 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 + # Only set when the spawn asked for a stderr pipe; inherited stderr leaves it 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 +140,7 @@ async def create_windows_process( command: str, args: list[str], env: dict[str, str] | None = None, - errlog: TextIO | None = sys.stderr, + errlog: TextIO | int | None = sys.stderr, cwd: Path | str | None = None, ) -> Process | FallbackProcess: """Creates a subprocess with Job Object support for tree termination. @@ -177,7 +180,7 @@ async def _create_windows_fallback_process( command: str, args: list[str], env: dict[str, str] | None = None, - errlog: TextIO | None = sys.stderr, + errlog: TextIO | int | None = sys.stderr, cwd: Path | str | None = None, ) -> FallbackProcess: """Spawns via subprocess.Popen and wraps it in FallbackProcess.""" diff --git a/tests/client/test_stdio.py b/tests/client/test_stdio.py index 91f829ff98..d29b281f4a 100644 --- a/tests/client/test_stdio.py +++ b/tests/client/test_stdio.py @@ -16,6 +16,7 @@ import sys from collections.abc import Callable from contextlib import AsyncExitStack, suppress +from io import StringIO from pathlib import Path from typing import TextIO, cast @@ -127,6 +128,10 @@ async def aclose(self) -> None: # Real async closes yield; keeps the fake honest and shutdown scheduling realistic. await anyio.lowlevel.checkpoint() + def close(self) -> None: + """Release the read end, as the kernel does when the process's pipe goes away.""" + self._inner.close() + class FakeProcess: """In-memory stand-in for the spawned server process. @@ -145,7 +150,15 @@ 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._stderr_send, stderr_receive = anyio.create_memory_object_stream[bytes](math.inf) + # Only read when errlog has no descriptor to inherit, so the client pipes stderr. + self.stderr = _FakeStdout( + stderr_receive, + eof_error=stderr_eof_error, + on_receive=lambda: None, + ) self._stdout_send, stdout_receive = anyio.create_memory_object_stream[bytes](math.inf) self.stdout = _FakeStdout( stdout_receive, @@ -178,10 +191,24 @@ 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. + + Closes both ends: unlike stdout, stderr goes unread whenever errlog owns a + descriptor, so nothing else would ever release the read end. + """ + self._stderr_send.close() + self.stderr.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 both output pipes, 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.""" @@ -975,6 +1002,71 @@ async def stubborn_terminate(proc: FakeProcess) -> None: # The fake "survived", so nothing ever EOF'd its stdout pipe; release it here # or its GC-time ResourceWarning would fail a later test. process.close_stdout() + process.close_stderr() + + +class _UnwritableErrlog(StringIO): + """A log sink that rejects writes, as a closed notebook cell's stream does.""" + + def __init__(self) -> None: + super().__init__() + self.attempted = anyio.Event() + + def write(self, s: str, /) -> int: + self.attempted.set() + raise ValueError("I/O operation on closed file") + + +@pytest.mark.anyio +async def test_a_closed_errlog_stops_stderr_forwarding_without_failing_the_session( + monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture +) -> None: + """Losing the log sink mid-session costs the diagnostics, never the session. + + A notebook cell that finishes closes the stream underneath the forwarder; the + server's own traffic has to survive that. + """ + errlog = _UnwritableErrlog() + process = FakeProcess(on_stdin_close=lambda: process.exit(0)) + install_fake_process(monkeypatch, process) + ping = JSONRPCRequest(jsonrpc="2.0", id=1, method="ping") + + with caplog.at_level(logging.DEBUG, logger="mcp.client.stdio"): + with anyio.fail_after(5): + async with stdio_client(FAKE_PARAMS, errlog=cast(TextIO, errlog)) as (read_stream, _): + await process.feed_stderr(b"server diagnostics no one will read\n") + await errlog.attempted.wait() + + # The failed write must not have disturbed the message path. + await process.feed(_line(ping)) + assert await _next_message(read_stream) == ping + + assert "the log stream was closed" in caplog.text + assert process.returncode == 0 + + +@pytest.mark.anyio +async def test_a_stderr_pipe_dying_with_the_server_ends_forwarding_quietly( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """A stderr pipe torn down with the process is shutdown noise, not an error. + + The proactor loop reports a hard-killed pipe as a reset rather than EOF, which + must not propagate out of the transport. + """ + errlog = StringIO() + process = FakeProcess( + on_stdin_close=lambda: process.exit(0), + stderr_eof_error=anyio.BrokenResourceError(), + ) + install_fake_process(monkeypatch, process) + + with anyio.fail_after(5): + async with stdio_client(FAKE_PARAMS, errlog=cast(TextIO, errlog)) as (_, _write): + await process.feed_stderr(b"last words\n") + + assert errlog.getvalue() == "last words\n" + assert process.returncode == 0 # --------------------------------------------------------------------------- diff --git a/tests/transports/stdio/test_lifecycle.py b/tests/transports/stdio/test_lifecycle.py index c9046927b9..302fce9018 100644 --- a/tests/transports/stdio/test_lifecycle.py +++ b/tests/transports/stdio/test_lifecycle.py @@ -14,6 +14,7 @@ import sys import threading from contextlib import AsyncExitStack +from io import StringIO from pathlib import Path from textwrap import dedent @@ -170,8 +171,9 @@ async def test_server_stderr_output_reaches_the_errlog_file( ) -> None: """What the server writes to stderr lands in the file passed as `errlog`. - The spawn hands over errlog's file descriptor as the child's stderr, so it must - be a real file -- an in-memory StringIO has no fileno. + A real file has a descriptor, so the spawn hands it straight to the child and no + forwarding task is involved. The descriptor-less case is covered by + test_server_stderr_output_reaches_an_errlog_without_a_file_descriptor. """ marker = "stdio-lifecycle stderr marker 4242" @@ -206,6 +208,45 @@ async def test_server_stderr_output_reaches_the_errlog_file( assert spawned_processes[0].returncode == 0 +@pytest.mark.anyio +async def test_server_stderr_output_reaches_an_errlog_without_a_file_descriptor( + spawned_processes: list[anyio.abc.Process | FallbackProcess], +) -> None: + """Server stderr reaches an `errlog` that a child process cannot inherit. + + Jupyter replaces sys.stderr with an ipykernel stream that has no descriptor to + hand over, which used to drop the server's diagnostics entirely (#156). StringIO + stands in for it here: same missing fileno, no notebook needed. + """ + marker = "stdio-lifecycle fd-less stderr marker 4243" + errlog = StringIO() + + async with AsyncExitStack() as stack: + sock, port = await open_liveness_listener() + stack.push_async_callback(sock.aclose) + + server = ( + f"import socket, sys\n" + f"s = socket.create_connection(('127.0.0.1', {port}))\n" + f"s.sendall(b'alive')\n" + f"sys.stderr.write({marker!r} + '\\n')\n" + f"sys.stderr.flush()\n" + f"sys.stdin.read()\n" + ) + params = StdioServerParameters(command=sys.executable, args=["-c", server]) + + # The bound covers one interpreter cold start on a loaded runner; a + # healthy run takes well under a second. + with anyio.fail_after(10.0): + async with stdio_client(params, errlog=errlog): + stream = await accept_alive(sock) + stack.push_async_callback(stream.aclose) + + # Shutdown drains the forwarder after the server dies, so the write has landed. + assert marker in errlog.getvalue() + assert spawned_processes[0].returncode == 0 + + @pytest.mark.skipif( not hasattr(os, "waitid"), reason="needs os.waitid(WNOWAIT); absent on Windows and macOS before 3.13" )