feat: add Jupyter notebook stderr support to stdio_client

- Add _is_jupyter_notebook() to detect Jupyter/IPython environments - Add _print_stderr() to format stderr with HTML in Jupyter (red color) - Add async _stderr_reader() to capture and display stderr - Pipe stderr (subprocess.PIPE) instead of redirecting to file - Update Windows process creation to support piped stderr - Extract stdout/stdin/stderr readers as module-level functions This enables server stderr output to be visible in Jupyter notebooks, addressing issue #156. Fixes#156

Author: dgenio

src/mcp/client/stdio/__init__.py

@@ -1,5 +1,6 @@
 importlogging
 importos
+importsubprocess
 importsys
 fromcontextlibimportasynccontextmanager
 frompathlibimportPath
@@ -48,6 +49,47 @@
 PROCESS_TERMINATION_TIMEOUT=2.0
 
 
+def_is_jupyter_notebook() ->bool:
+"""
+ Detect if running in a Jupyter notebook or IPython environment.
+
+ Returns:
+ bool: True if running in Jupyter/IPython, False otherwise
+ """
+try:
+fromIPythonimportget_ipython# type: ignore[import-not-found]
+
+ipython=get_ipython() # type: ignore[no-untyped-call]
+returnipythonisnotNoneandipython.__class__.__name__in ("ZMQInteractiveShell", "TerminalInteractiveShell")
+exceptImportError:
+returnFalse
+
+
+def_print_stderr(line: str, errlog: TextIO) ->None:
+"""
+ Print stderr output, using IPython's display system if in Jupyter notebook.
+
+ Args:
+ line: The line to print
+ errlog: The fallback TextIO stream (used when not in Jupyter)
+ """
+if_is_jupyter_notebook():
+try:
+fromIPython.displayimportHTML, display# type: ignore[import-not-found]
+
+# Use IPython's display system with red color for stderr
+# This ensures proper rendering in Jupyter notebooks
+display(HTML(f'<pre style="color: red;">{line}</pre>')) # type: ignore[no-untyped-call]
+exceptException:
+# If IPython display fails, fall back to regular print
+# Log the error but continue (non-critical)
+logger.debug("Failed to use IPython display for stderr, falling back to print", exc_info=True)
+print(line, file=errlog)
+else:
+# Not in Jupyter, use standard stderr redirection
+print(line, file=errlog)
+
+
 defget_default_environment() ->dict[str, str]:
 """
  Returns a default environment object including only environment variables deemed
@@ -102,11 +144,121 @@ class StdioServerParameters(BaseModel):
  """
 
 
+asyncdef_stdout_reader(
+process: Process|FallbackProcess,
+read_stream_writer: MemoryObjectSendStream[SessionMessage|Exception],
+encoding: str,
+encoding_error_handler: str,
+):
+"""Read stdout from the process and parse JSONRPC messages."""
+assertprocess.stdout, "Opened process is missing stdout"
+
+try:
+asyncwithread_stream_writer:
+buffer=""
+asyncforchunkinTextReceiveStream(
+process.stdout,
+encoding=encoding,
+errors=encoding_error_handler,
+ ):
+lines= (buffer+chunk).split("\n")
+buffer=lines.pop()
+
+forlineinlines:
+try:
+message=types.JSONRPCMessage.model_validate_json(line)
+exceptExceptionasexc: # pragma: no cover
+logger.exception("Failed to parse JSONRPC message from server")
+awaitread_stream_writer.send(exc)
+continue
+
+session_message=SessionMessage(message)
+awaitread_stream_writer.send(session_message)
+exceptanyio.ClosedResourceError: # pragma: no cover
+awaitanyio.lowlevel.checkpoint()
+
+
+asyncdef_stdin_writer(
+process: Process|FallbackProcess,
+write_stream_reader: MemoryObjectReceiveStream[SessionMessage],
+encoding: str,
+encoding_error_handler: str,
+):
+"""Write session messages to the process stdin."""
+assertprocess.stdin, "Opened process is missing stdin"
+
+try:
+asyncwithwrite_stream_reader:
+asyncforsession_messageinwrite_stream_reader:
+json=session_message.message.model_dump_json(by_alias=True, exclude_none=True)
+awaitprocess.stdin.send(
+ (json+"\n").encode(
+encoding=encoding,
+errors=encoding_error_handler,
+ )
+ )
+exceptanyio.ClosedResourceError: # pragma: no cover
+awaitanyio.lowlevel.checkpoint()
+
+
+asyncdef_stderr_reader(
+process: Process|FallbackProcess,
+errlog: TextIO,
+encoding: str,
+encoding_error_handler: str,
+):
+"""Read stderr from the process and display it appropriately."""
+ifnotprocess.stderr:
+return
+
+try:
+buffer=""
+asyncforchunkinTextReceiveStream(
+process.stderr,
+encoding=encoding,
+errors=encoding_error_handler,
+ ):
+lines= (buffer+chunk).split("\n")
+buffer=lines.pop()
+
+forlineinlines:
+ifline.strip(): # Only print non-empty lines
+try:
+_print_stderr(line, errlog)
+exceptException:
+# Log errors but continue (non-critical)
+logger.debug("Failed to print stderr line", exc_info=True)
+
+# Print any remaining buffer content
+ifbuffer.strip():
+try:
+_print_stderr(buffer, errlog)
+exceptException:
+logger.debug("Failed to print final stderr buffer", exc_info=True)
+exceptanyio.ClosedResourceError: # pragma: no cover
+awaitanyio.lowlevel.checkpoint()
+exceptException:
+# Log errors but continue (non-critical)
+logger.debug("Error reading stderr", exc_info=True)
+
+
 @asynccontextmanager
 asyncdefstdio_client(server: StdioServerParameters, errlog: TextIO=sys.stderr):
 """
  Client transport for stdio: this will connect to a server by spawning a
  process and communicating with it over stdin/stdout.
+
+ This function automatically handles stderr output in a way that is compatible
+ with Jupyter notebook environments. When running in Jupyter, stderr output
+ is displayed using IPython's display system with red color formatting.
+ When not in Jupyter, stderr is redirected to the provided errlog stream
+ (defaults to sys.stderr).
+
+ Args:
+ server: Parameters for the server process to spawn
+ errlog: TextIO stream for stderr output when not in Jupyter (defaults to sys.stderr).
+ This parameter is kept for backward compatibility but may be ignored
+ when running in Jupyter notebook environments.
  """
 read_stream: MemoryObjectReceiveStream[SessionMessage|Exception]
 read_stream_writer: MemoryObjectSendStream[SessionMessage|Exception]
@@ -136,55 +288,14 @@ async def stdio_client(server: StdioServerParameters, errlog: TextIO = sys.stder
 awaitwrite_stream_reader.aclose()
 raise
 
-asyncdefstdout_reader():
-assertprocess.stdout, "Opened process is missing stdout"
-
-try:
-asyncwithread_stream_writer:
-buffer=""
-asyncforchunkinTextReceiveStream(
-process.stdout,
-encoding=server.encoding,
-errors=server.encoding_error_handler,
- ):
-lines= (buffer+chunk).split("\n")
-buffer=lines.pop()
-
-forlineinlines:
-try:
-message=types.JSONRPCMessage.model_validate_json(line)
-exceptExceptionasexc: # pragma: no cover
-logger.exception("Failed to parse JSONRPC message from server")
-awaitread_stream_writer.send(exc)
-continue
-
-session_message=SessionMessage(message)
-awaitread_stream_writer.send(session_message)
-exceptanyio.ClosedResourceError: # pragma: no cover
-awaitanyio.lowlevel.checkpoint()
-
-asyncdefstdin_writer():
-assertprocess.stdin, "Opened process is missing stdin"
-
-try:
-asyncwithwrite_stream_reader:
-asyncforsession_messageinwrite_stream_reader:
-json=session_message.message.model_dump_json(by_alias=True, exclude_none=True)
-awaitprocess.stdin.send(
- (json+"\n").encode(
-encoding=server.encoding,
-errors=server.encoding_error_handler,
- )
- )
-exceptanyio.ClosedResourceError: # pragma: no cover
-awaitanyio.lowlevel.checkpoint()
-
 asyncwith (
 anyio.create_task_group() astg,
 process,
  ):
-tg.start_soon(stdout_reader)
-tg.start_soon(stdin_writer)
+tg.start_soon(_stdout_reader, process, read_stream_writer, server.encoding, server.encoding_error_handler)
+tg.start_soon(_stdin_writer, process, write_stream_reader, server.encoding, server.encoding_error_handler)
+ifprocess.stderr:
+tg.start_soon(_stderr_reader, process, errlog, server.encoding, server.encoding_error_handler)
 try:
 yieldread_stream, write_stream
 finally:
@@ -244,14 +355,19 @@ async def _create_platform_compatible_process(
 
  Unix: Creates process in a new session/process group for killpg support
  Windows: Creates process in a Job Object for reliable child termination
+
+ Note: stderr is piped (not redirected) to allow async reading for Jupyter
+ notebook compatibility. The errlog parameter is kept for backward compatibility
+ but is only used when not in Jupyter environments.
  """
 ifsys.platform=="win32": # pragma: no cover
-process=awaitcreate_windows_process(command, args, env, errlog, cwd)
+process=awaitcreate_windows_process(command, args, env, errlog, cwd, pipe_stderr=True)
 else:
+# Pipe stderr instead of redirecting to allow async reading
 process=awaitanyio.open_process(
  [command, *args],
 env=env,
-stderr=errlog,
+stderr=subprocess.PIPE,
 cwd=cwd,
 start_new_session=True,
  ) # pragma: no cover

src/mcp/os/win32/utilities.py

@@ -70,7 +70,7 @@ class FallbackProcess:
  A fallback process wrapper for Windows to handle async I/O
  when using subprocess.Popen, which provides sync-only FileIO objects.
 
- This wraps stdinand stdout into async-compatible
+ This wraps stdin, stdout, and stderr into async-compatible
  streams (FileReadStream, FileWriteStream),
  so that MCP clients expecting async streams can work properly.
  """
@@ -79,10 +79,12 @@ def __init__(self, popen_obj: subprocess.Popen[bytes]):
 self.popen: subprocess.Popen[bytes] =popen_obj
 self.stdin_raw=popen_obj.stdin# type: ignore[assignment]
 self.stdout_raw=popen_obj.stdout# type: ignore[assignment]
-self.stderr=popen_obj.stderr# type: ignore[assignment]
+self.stderr_raw=popen_obj.stderr# type: ignore[assignment]
 
 self.stdin=FileWriteStream(cast(BinaryIO, self.stdin_raw)) ifself.stdin_rawelseNone
 self.stdout=FileReadStream(cast(BinaryIO, self.stdout_raw)) ifself.stdout_rawelseNone
+# Wrap stderr in async stream if it's piped (for Jupyter compatibility)
+self.stderr=FileReadStream(cast(BinaryIO, self.stderr_raw)) ifself.stderr_rawelseNone
 
 asyncdef__aenter__(self):
 """Support async context manager entry."""
@@ -103,12 +105,14 @@ async def __aexit__(
 awaitself.stdin.aclose()
 ifself.stdout:
 awaitself.stdout.aclose()
+ifself.stderr:
+awaitself.stderr.aclose()
 ifself.stdin_raw:
 self.stdin_raw.close()
 ifself.stdout_raw:
 self.stdout_raw.close()
-ifself.stderr:
-self.stderr.close()
+ifself.stderr_raw:
+self.stderr_raw.close()
 
 asyncdefwait(self):
 """Async wait for process completion."""
@@ -139,6 +143,7 @@ async def create_windows_process(
 env: dict[str, str] |None=None,
 errlog: TextIO|None=sys.stderr,
 cwd: Path|str|None=None,
+pipe_stderr: bool=False,
 ) ->Process|FallbackProcess:
 """
  Creates a subprocess in a Windows-compatible way with Job Object support.
@@ -155,15 +160,20 @@ async def create_windows_process(
  command (str): The executable to run
  args (list[str]): List of command line arguments
  env (dict[str, str] | None): Environment variables
- errlog (TextIO | None): Where to send stderr output (defaults to sys.stderr)
+ errlog (TextIO | None): Where to send stderr output (defaults to sys.stderr).
+ Only used when pipe_stderr is False.
  cwd (Path | str | None): Working directory for the subprocess
+ pipe_stderr (bool): If True, pipe stderr instead of redirecting to errlog.
+ This allows async reading of stderr for Jupyter compatibility.
 
  Returns:
  Process | FallbackProcess: Async-compatible subprocess with stdin and stdout streams
  """
 job=_create_job_object()
 process=None
 
+stderr_target=subprocess.PIPEifpipe_stderrelseerrlog
+
 try:
 # First try using anyio with Windows-specific flags to hide console window
 process=awaitanyio.open_process(
@@ -173,18 +183,18 @@ async def create_windows_process(
 creationflags=subprocess.CREATE_NO_WINDOW# type: ignore
 ifhasattr(subprocess, "CREATE_NO_WINDOW")
 else0,
-stderr=errlog,
+stderr=stderr_target,
 cwd=cwd,
  )
 exceptNotImplementedError:
 # If Windows doesn't support async subprocess creation, use fallback
-process=await_create_windows_fallback_process(command, args, env, errlog, cwd)
+process=await_create_windows_fallback_process(command, args, env, errlog, cwd, pipe_stderr=pipe_stderr)
 exceptException:
 # Try again without creation flags
 process=awaitanyio.open_process(
  [command, *args],
 env=env,
-stderr=errlog,
+stderr=stderr_target,
 cwd=cwd,
  )
 
@@ -198,19 +208,30 @@ async def _create_windows_fallback_process(
 env: dict[str, str] |None=None,
 errlog: TextIO|None=sys.stderr,
 cwd: Path|str|None=None,
+pipe_stderr: bool=False,
 ) ->FallbackProcess:
 """
  Create a subprocess using subprocess.Popen as a fallback when anyio fails.
 
  This function wraps the sync subprocess.Popen in an async-compatible interface.
+
+ Args:
+ command: The executable to run
+ args: List of command line arguments
+ env: Environment variables
+ errlog: Where to send stderr output (only used when pipe_stderr is False)
+ cwd: Working directory for the subprocess
+ pipe_stderr: If True, pipe stderr instead of redirecting to errlog
  """
+stderr_target=subprocess.PIPEifpipe_stderrelseerrlog
+
 try:
 # Try launching with creationflags to avoid opening a new console window
 popen_obj=subprocess.Popen(
  [command, *args],
 stdin=subprocess.PIPE,
 stdout=subprocess.PIPE,
-stderr=errlog,
+stderr=stderr_target,
 env=env,
 cwd=cwd,
 bufsize=0, # Unbuffered output
@@ -222,7 +243,7 @@ async def _create_windows_fallback_process(
  [command, *args],
 stdin=subprocess.PIPE,
 stdout=subprocess.PIPE,
-stderr=errlog,
+stderr=stderr_target,
 env=env,
 cwd=cwd,
 bufsize=0,