fix: route subprocess stderr via reader task for Jupyter support (#156)

In Jupyter environments, stderr from subprocesses isn't directly visible.
To fix this, we pipe stderr via subprocess.PIPE and use a background reader
task that detects Jupyter (via is_jupyter()) and prints the output with
ANSI red coloring. For standard environments, the output is routed to
sys.stderr as before.

Also updated create_windows_process to accept int for errlog, allowing
subprocess.PIPE.

Closes #156

Author: Ashutosh0x

src/mcp/client/stdio.py

@@ -1,5 +1,6 @@
 import logging
 import os
+import subprocess
 import sys
 from contextlib import asynccontextmanager
 from pathlib import Path
@@ -20,6 +21,7 @@
     get_windows_executable_command,
     terminate_windows_process_tree,
 )
+from mcp.shared.jupyter import is_jupyter
 from mcp.shared.message import SessionMessage
 
 logger = logging.getLogger(__name__)
@@ -118,12 +120,13 @@ async def stdio_client(server: StdioServerParameters, errlog: TextIO = sys.stder
     try:
         command = _get_executable_command(server.command)
 
-        # Open process with stderr piped for capture
+        # Pipe stderr so we can route it through a reader task.
+        # This enables Jupyter-compatible stderr output (#156).
         process = await _create_platform_compatible_process(
             command=command,
             args=server.args,
             env=({**get_default_environment(), **server.env} if server.env is not None else get_default_environment()),
-            errlog=errlog,
+            errlog=subprocess.PIPE,
             cwd=server.cwd,
         )
     except OSError:
@@ -177,9 +180,28 @@ async def stdin_writer():
         except anyio.ClosedResourceError:  # pragma: no cover
             await anyio.lowlevel.checkpoint()
 
+    async def stderr_reader():
+        """Read stderr from the subprocess and route to errlog or Jupyter output."""
+        if process.stderr is None:  # pragma: no cover
+            return
+        try:
+            async for chunk in TextReceiveStream(
+                process.stderr,
+                encoding=server.encoding,
+                errors=server.encoding_error_handler,
+            ):
+                if is_jupyter():
+                    # In Jupyter, stderr isn't visible — use print() with ANSI red
+                    print(f"\033[91m{chunk}\033[0m", end="", flush=True)
+                else:
+                    print(chunk, file=errlog, end="", flush=True)
+        except anyio.ClosedResourceError:
+            await anyio.lowlevel.checkpoint()
+
     async with anyio.create_task_group() as tg, process:
         tg.start_soon(stdout_reader)
         tg.start_soon(stdin_writer)
+        tg.start_soon(stderr_reader)
         try:
             yield read_stream, write_stream
         finally:
@@ -230,7 +252,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,
 ):
     """Creates a subprocess in a platform-compatible way.

src/mcp/os/win32/utilities.py

@@ -133,7 +133,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 in a Windows-compatible way with Job Object support.

src/mcp/shared/jupyter.py

@@ -0,0 +1,12 @@
+def is_jupyter() -> bool:
+    """Check if we are running in a Jupyter notebook environment."""
+    try:
+        shell = get_ipython().__class__.__name__  # type: ignore
+        if shell == "ZMQInteractiveShell":
+            return True  # Jupyter notebook or qtconsole
+        elif shell == "TerminalInteractiveShell":
+            return False  # Terminal running IPython
+        else:
+            return False  # Other type (?)
+    except NameError:
+        return False  # Probably standard Python interpreter

tests/issues/test_156_jupyter_detection.py

@@ -0,0 +1,35 @@
+"""Tests for the is_jupyter() helper used by issue #156."""
+
+import builtins
+from unittest.mock import MagicMock
+
+from mcp.shared.jupyter import is_jupyter
+
+
+def test_is_jupyter_false_in_standard_python():
+    """In a standard Python interpreter, is_jupyter() should return False."""
+    assert is_jupyter() is False
+
+
+def test_is_jupyter_true_in_zmq_shell():
+    """When get_ipython() returns a ZMQInteractiveShell, we're in Jupyter."""
+    mock_ipython = MagicMock()
+    mock_ipython.__class__.__name__ = "ZMQInteractiveShell"
+
+    builtins.get_ipython = lambda: mock_ipython  # type: ignore[attr-defined]
+    try:
+        assert is_jupyter() is True
+    finally:
+        del builtins.get_ipython  # type: ignore[attr-defined]
+
+
+def test_is_jupyter_false_in_terminal_ipython():
+    """When get_ipython() returns TerminalInteractiveShell, we're in IPython (not Jupyter)."""
+    mock_ipython = MagicMock()
+    mock_ipython.__class__.__name__ = "TerminalInteractiveShell"
+
+    builtins.get_ipython = lambda: mock_ipython  # type: ignore[attr-defined]
+    try:
+        assert is_jupyter() is False
+    finally:
+        del builtins.get_ipython  # type: ignore[attr-defined]

tests/issues/test_156_stdio_stderr.py

@@ -0,0 +1,78 @@
+"""Regression test for issue #156: Jupyter Notebook stderr logging.
+
+When running in Jupyter, stderr from subprocess servers isn't visible because
+Jupyter doesn't display stderr output directly.  The fix pipes stderr via
+subprocess.PIPE and adds a reader task that detects Jupyter and uses print()
+with ANSI red colouring.
+"""
+
+import sys
+import textwrap
+
+import anyio
+import pytest
+
+from mcp.client.stdio import StdioServerParameters, stdio_client
+
+
+# A minimal MCP-like server that writes to stderr and then exits.
+SERVER_SCRIPT = textwrap.dedent("""\
+    import sys
+    sys.stderr.write("hello from stderr\\n")
+    sys.stderr.flush()
+    # Read stdin until EOF so the process doesn't exit before client reads stderr
+    sys.stdin.read()
+""")
+
+
+@pytest.mark.anyio
+async def test_stderr_is_captured(capsys: pytest.CaptureFixture[str]) -> None:
+    """Verify that subprocess stderr is captured and printed to errlog (sys.stderr)."""
+    from unittest.mock import patch
+
+    params = StdioServerParameters(command=sys.executable, args=["-c", SERVER_SCRIPT])
+
+    # Force is_jupyter=False so we use the standard errlog path
+    # Pass sys.stderr explicitly so we use the capsys-patched stderr
+    with patch("mcp.client.stdio.is_jupyter", return_value=False), anyio.fail_after(10):
+        async with stdio_client(params, errlog=sys.stderr) as (_read, _write):
+            # Give the stderr_reader task time to process
+            await anyio.sleep(0.5)
+
+    captured = capsys.readouterr()
+    # verify it went to stderr
+    assert "hello from stderr" in captured.err
+
+
+@pytest.mark.anyio
+async def test_stderr_is_routed_to_errlog() -> None:
+    """Verify that subprocess stderr is written to the provided explicit errlog."""
+    import io
+    from unittest.mock import patch
+
+    errlog = io.StringIO()
+    params = StdioServerParameters(command=sys.executable, args=["-c", SERVER_SCRIPT])
+
+    with patch("mcp.client.stdio.is_jupyter", return_value=False), anyio.fail_after(10):
+        async with stdio_client(params, errlog=errlog) as (_read, _write):
+            await anyio.sleep(0.5)
+
+    assert "hello from stderr" in errlog.getvalue()
+
+
+@pytest.mark.anyio
+async def test_stderr_is_printed_with_color_in_jupyter(capsys: pytest.CaptureFixture[str]) -> None:
+    """Verify that subprocess stderr is printed with ANSI red in Jupyter."""
+    from unittest.mock import patch
+
+    params = StdioServerParameters(command=sys.executable, args=["-c", SERVER_SCRIPT])
+
+    # Force is_jupyter=True so we use the print() path
+    with patch("mcp.client.stdio.is_jupyter", return_value=True), anyio.fail_after(10):
+        async with stdio_client(params) as (_read, _write):
+            await anyio.sleep(0.5)
+
+    captured = capsys.readouterr()
+    # print() goes to stdout by default
+    assert "\033[91mhello from stderr" in captured.out
+