From e9c5fd2f978f8fdc94832270a10dbaf5c45c7b4f Mon Sep 17 00:00:00 2001
From: BabyChrist666 <istiaqrafi39@gmail.com>
Date: Sun, 8 Feb 2026 19:43:54 -0500
Subject: [PATCH 1/7] feat: support stderr logging in Jupyter notebook
 environments

Fixes #156

In Jupyter notebook environments, passing sys.stderr directly to subprocess
doesn't work as expected - the stderr output is lost and users have no
visibility into server errors or crashes.

This change:
- Adds _is_jupyter_environment() to detect when running in Jupyter/IPython
- In Jupyter mode, captures stderr as a pipe and reads it asynchronously
- Prints stderr output with red ANSI color for visibility in notebooks
- In non-Jupyter environments, behavior remains unchanged (direct stderr)

The fix ensures that MCP server error messages and crash logs are visible
in Jupyter notebooks, making debugging much easier.

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
---
 src/mcp/client/stdio.py       | 94 +++++++++++++++++++++++++++++++++--
 src/mcp/os/win32/utilities.py | 21 +++++---
 tests/client/test_stdio.py    | 79 +++++++++++++++++++++++++++++
 3 files changed, 182 insertions(+), 12 deletions(-)

diff --git a/src/mcp/client/stdio.py b/src/mcp/client/stdio.py
index 605c5ea24..f8f4baf5a 100644
--- a/src/mcp/client/stdio.py
+++ b/src/mcp/client/stdio.py
@@ -1,9 +1,10 @@
 import logging
 import os
+import subprocess
 import sys
 from contextlib import asynccontextmanager
 from pathlib import Path
-from typing import Literal, TextIO
+from typing import Callable, Literal, TextIO
 
 import anyio
 import anyio.lowlevel
@@ -24,6 +25,32 @@
 
 logger = logging.getLogger(__name__)
 
+
+def _is_jupyter_environment() -> bool:
+    """Detect if code is running in a Jupyter notebook environment.
+
+    In Jupyter environments, sys.stderr doesn't work as expected when passed
+    to subprocess, so we need to handle stderr differently.
+
+    Returns:
+        bool: True if running in Jupyter/IPython notebook environment
+    """
+    try:
+        # Check for IPython kernel
+        from IPython import get_ipython
+
+        ipython = get_ipython()
+        if ipython is not None:
+            # Check if it's a notebook kernel (not just IPython terminal)
+            if "IPKernelApp" in ipython.config:
+                return True
+            # Also check for ZMQInteractiveShell which indicates notebook
+            if ipython.__class__.__name__ == "ZMQInteractiveShell":
+                return True
+    except (ImportError, AttributeError):
+        pass
+    return False
+
 # Environment variables to inherit by default
 DEFAULT_INHERITED_ENV_VARS = (
     [
@@ -105,6 +132,12 @@ class StdioServerParameters(BaseModel):
 async def stdio_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.
+
+    Args:
+        server: Parameters for the server process
+        errlog: TextIO stream for stderr output. In Jupyter environments,
+                stderr is captured and printed to work around Jupyter's
+                limitations with subprocess stderr handling.
     """
     read_stream: MemoryObjectReceiveStream[SessionMessage | Exception]
     read_stream_writer: MemoryObjectSendStream[SessionMessage | Exception]
@@ -115,16 +148,20 @@ async def stdio_client(server: StdioServerParameters, errlog: TextIO = sys.stder
     read_stream_writer, read_stream = anyio.create_memory_object_stream(0)
     write_stream, write_stream_reader = anyio.create_memory_object_stream(0)
 
+    # Detect Jupyter environment for stderr handling
+    is_jupyter = _is_jupyter_environment()
+
     try:
         command = _get_executable_command(server.command)
 
-        # Open process with stderr piped for capture
+        # Open process with stderr handling based on environment
         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,
             cwd=server.cwd,
+            capture_stderr=is_jupyter,
         )
     except OSError:
         # Clean up streams if process creation fails
@@ -177,9 +214,41 @@ async def stdin_writer():
         except anyio.ClosedResourceError:  # pragma: no cover
             await anyio.lowlevel.checkpoint()
 
+    async def stderr_reader():
+        """Read stderr from the process and output it appropriately.
+
+        In Jupyter environments, stderr is captured as a pipe and printed
+        to make it visible in the notebook output. In normal environments,
+        stderr is passed directly to sys.stderr.
+
+        See: https://github.com/modelcontextprotocol/python-sdk/issues/156
+        """
+        if not process.stderr:
+            return
+
+        try:
+            async for chunk in TextReceiveStream(
+                process.stderr,
+                encoding=server.encoding,
+                errors=server.encoding_error_handler,
+            ):
+                # In Jupyter, print to stdout with red color for visibility
+                # In normal environments, write to the provided errlog
+                if is_jupyter:
+                    # Use ANSI red color for stderr in Jupyter
+                    print(f"\033[91m{chunk}\033[0m", end="", flush=True)
+                else:
+                    errlog.write(chunk)
+                    errlog.flush()
+        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)
+        # Only start stderr reader if we're capturing stderr (Jupyter mode)
+        if is_jupyter and process.stderr:
+            tg.start_soon(stderr_reader)
         try:
             yield read_stream, write_stream
         finally:
@@ -232,19 +301,36 @@ async def _create_platform_compatible_process(
     env: dict[str, str] | None = None,
     errlog: TextIO = sys.stderr,
     cwd: Path | str | None = None,
+    capture_stderr: bool = False,
 ):
     """Creates a subprocess in a platform-compatible way.
 
     Unix: Creates process in a new session/process group for killpg support
     Windows: Creates process in a Job Object for reliable child termination
+
+    Args:
+        command: The executable command to run
+        args: Command line arguments
+        env: Environment variables for the process
+        errlog: TextIO stream for stderr (used when capture_stderr=False)
+        cwd: Working directory for the process
+        capture_stderr: If True, stderr is captured as a pipe for async reading.
+                       This is needed for Jupyter environments where passing
+                       sys.stderr directly doesn't work properly.
+
+    Returns:
+        Process with stdin, stdout, and optionally stderr streams
     """
+    # Determine stderr handling: PIPE for capture, or redirect to errlog
+    stderr_target = subprocess.PIPE if capture_stderr else errlog
+
     if sys.platform == "win32":  # pragma: no cover
-        process = await create_windows_process(command, args, env, errlog, cwd)
+        process = await create_windows_process(command, args, env, stderr_target, cwd)
     else:  # pragma: lax no cover
         process = await anyio.open_process(
             [command, *args],
             env=env,
-            stderr=errlog,
+            stderr=stderr_target,
             cwd=cwd,
             start_new_session=True,
         )
diff --git a/src/mcp/os/win32/utilities.py b/src/mcp/os/win32/utilities.py
index fa4e4b399..5be39864f 100644
--- a/src/mcp/os/win32/utilities.py
+++ b/src/mcp/os/win32/utilities.py
@@ -5,7 +5,7 @@
 import subprocess
 import sys
 from pathlib import Path
-from typing import BinaryIO, TextIO, cast
+from typing import BinaryIO, TextIO, Union, cast
 
 import anyio
 from anyio import to_thread
@@ -66,7 +66,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 stdin and stdout into async-compatible
+    This wraps stdin, stdout, and optionally stderr into async-compatible
     streams (FileReadStream, FileWriteStream),
     so that MCP clients expecting async streams can work properly.
     """
@@ -75,10 +75,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)) if self.stdin_raw else None
         self.stdout = FileReadStream(cast(BinaryIO, self.stdout_raw)) if self.stdout_raw else None
+        # Wrap stderr as async stream if it was captured as PIPE
+        self.stderr = FileReadStream(cast(BinaryIO, self.stderr_raw)) if self.stderr_raw else None
 
     async def __aenter__(self):
         """Support async context manager entry."""
@@ -99,12 +101,14 @@ async def __aexit__(
             await self.stdin.aclose()
         if self.stdout:
             await self.stdout.aclose()
+        if self.stderr:
+            await self.stderr.aclose()
         if self.stdin_raw:
             self.stdin_raw.close()
         if self.stdout_raw:
             self.stdout_raw.close()
-        if self.stderr:
-            self.stderr.close()
+        if self.stderr_raw:
+            self.stderr_raw.close()
 
     async def wait(self):
         """Async wait for process completion."""
@@ -133,7 +137,7 @@ async def create_windows_process(
     command: str,
     args: list[str],
     env: dict[str, str] | None = None,
-    errlog: TextIO | None = sys.stderr,
+    errlog: Union[TextIO, int, None] = sys.stderr,
     cwd: Path | str | None = None,
 ) -> Process | FallbackProcess:
     """Creates a subprocess in a Windows-compatible way with Job Object support.
@@ -150,7 +154,8 @@ 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: Where to send stderr output. Can be a TextIO stream (like sys.stderr),
+                subprocess.PIPE (-1) for capturing stderr, or None.
         cwd (Path | str | None): Working directory for the subprocess
 
     Returns:
@@ -191,7 +196,7 @@ async def _create_windows_fallback_process(
     command: str,
     args: list[str],
     env: dict[str, str] | None = None,
-    errlog: TextIO | None = sys.stderr,
+    errlog: Union[TextIO, int, None] = sys.stderr,
     cwd: Path | str | None = None,
 ) -> FallbackProcess:
     """Create a subprocess using subprocess.Popen as a fallback when anyio fails.
diff --git a/tests/client/test_stdio.py b/tests/client/test_stdio.py
index f70c24eee..b7523d127 100644
--- a/tests/client/test_stdio.py
+++ b/tests/client/test_stdio.py
@@ -13,6 +13,7 @@
 from mcp.client.stdio import (
     StdioServerParameters,
     _create_platform_compatible_process,
+    _is_jupyter_environment,
     _terminate_process_tree,
     stdio_client,
 )
@@ -620,3 +621,81 @@ def sigterm_handler(signum, frame):
         f"stdio_client cleanup took {elapsed:.1f} seconds for stdin-ignoring process. "
         f"Expected between 2-4 seconds (2s stdin timeout + termination time)."
     )
+
+
+class TestJupyterStderrSupport:
+    """Tests for Jupyter notebook stderr logging support.
+
+    See: https://github.com/modelcontextprotocol/python-sdk/issues/156
+    """
+
+    def test_jupyter_detection_not_in_jupyter(self):
+        """Test that _is_jupyter_environment returns False when not in Jupyter."""
+        # In a normal Python environment (like pytest), this should return False
+        result = _is_jupyter_environment()
+        assert result is False, "Should not detect Jupyter in normal Python environment"
+
+    def test_jupyter_detection_handles_missing_ipython(self):
+        """Test that _is_jupyter_environment handles missing IPython gracefully."""
+        # This test verifies the ImportError handling works
+        # by calling the function when IPython may or may not be installed
+        result = _is_jupyter_environment()
+        # Should return False (not crash) regardless of IPython availability
+        assert isinstance(result, bool)
+
+    @pytest.mark.anyio
+    async def test_stderr_captured_in_process(self):
+        """Test that stderr output from a subprocess can be captured."""
+        # Create a script that writes to stderr
+        script = textwrap.dedent(
+            '''
+            import sys
+            sys.stderr.write("test error message\\n")
+            sys.stderr.flush()
+            # Exit immediately
+            sys.exit(0)
+            '''
+        )
+
+        server_params = StdioServerParameters(
+            command=sys.executable,
+            args=["-c", script],
+        )
+
+        # The stdio_client should handle this without hanging
+        with anyio.move_on_after(3.0) as cancel_scope:
+            async with stdio_client(server_params) as (read_stream, write_stream):
+                await anyio.sleep(0.5)  # Give process time to write and exit
+
+        assert not cancel_scope.cancelled_caught, "stdio_client should not hang on stderr output"
+
+    @pytest.mark.anyio
+    async def test_stderr_with_continuous_output(self):
+        """Test that continuous stderr output doesn't block the client."""
+        # Create a script that writes to stderr continuously then exits
+        script = textwrap.dedent(
+            '''
+            import sys
+            import time
+
+            for i in range(5):
+                sys.stderr.write(f"stderr line {i}\\n")
+                sys.stderr.flush()
+                time.sleep(0.1)
+
+            # Exit after writing
+            sys.exit(0)
+            '''
+        )
+
+        server_params = StdioServerParameters(
+            command=sys.executable,
+            args=["-c", script],
+        )
+
+        # The client should handle continuous stderr without blocking
+        with anyio.move_on_after(5.0) as cancel_scope:
+            async with stdio_client(server_params) as (read_stream, write_stream):
+                await anyio.sleep(1.0)  # Wait for stderr output
+
+        assert not cancel_scope.cancelled_caught, "stdio_client should handle continuous stderr output"

From 49e25460e137e1087cf290c9d5a41dac9b43642d Mon Sep 17 00:00:00 2001
From: BabyChrist666 <istiaqrafi39@gmail.com>
Date: Sun, 15 Feb 2026 00:49:48 -0500
Subject: [PATCH 2/7] Fix pre-commit formatting and add test coverage for
 Jupyter stderr support

- Add missing blank line after _is_jupyter_environment function (ruff E302)
- Change triple-quoted strings from ''' to """ in tests (ruff Q000)
- Remove unused Callable import
- Remove dead code branch in stderr_reader (non-Jupyter path was unreachable)
- Add tests for _is_jupyter_environment with mocked IPython (IPKernelApp, ZMQInteractiveShell)
- Add tests for stderr_reader in Jupyter mode with mocked _is_jupyter_environment

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
---
 src/mcp/client/stdio.py    |  22 +++-----
 tests/client/test_stdio.py | 102 +++++++++++++++++++++++++++++++++++--
 2 files changed, 106 insertions(+), 18 deletions(-)

diff --git a/src/mcp/client/stdio.py b/src/mcp/client/stdio.py
index f8f4baf5a..3b5c45d0e 100644
--- a/src/mcp/client/stdio.py
+++ b/src/mcp/client/stdio.py
@@ -4,7 +4,7 @@
 import sys
 from contextlib import asynccontextmanager
 from pathlib import Path
-from typing import Callable, Literal, TextIO
+from typing import Literal, TextIO
 
 import anyio
 import anyio.lowlevel
@@ -51,6 +51,7 @@ def _is_jupyter_environment() -> bool:
         pass
     return False
 
+
 # Environment variables to inherit by default
 DEFAULT_INHERITED_ENV_VARS = (
     [
@@ -215,16 +216,15 @@ async def stdin_writer():
             await anyio.lowlevel.checkpoint()
 
     async def stderr_reader():
-        """Read stderr from the process and output it appropriately.
+        """Read stderr from the process and print it to notebook output.
 
         In Jupyter environments, stderr is captured as a pipe and printed
-        to make it visible in the notebook output. In normal environments,
-        stderr is passed directly to sys.stderr.
+        to make it visible in the notebook output.
 
         See: https://github.com/modelcontextprotocol/python-sdk/issues/156
         """
         if not process.stderr:
-            return
+            return  # pragma: no cover
 
         try:
             async for chunk in TextReceiveStream(
@@ -232,15 +232,9 @@ async def stderr_reader():
                 encoding=server.encoding,
                 errors=server.encoding_error_handler,
             ):
-                # In Jupyter, print to stdout with red color for visibility
-                # In normal environments, write to the provided errlog
-                if is_jupyter:
-                    # Use ANSI red color for stderr in Jupyter
-                    print(f"\033[91m{chunk}\033[0m", end="", flush=True)
-                else:
-                    errlog.write(chunk)
-                    errlog.flush()
-        except anyio.ClosedResourceError:
+                # Use ANSI red color for stderr visibility in Jupyter
+                print(f"\033[91m{chunk}\033[0m", end="", flush=True)
+        except anyio.ClosedResourceError:  # pragma: no cover
             await anyio.lowlevel.checkpoint()
 
     async with anyio.create_task_group() as tg, process:
diff --git a/tests/client/test_stdio.py b/tests/client/test_stdio.py
index b7523d127..4b80c7476 100644
--- a/tests/client/test_stdio.py
+++ b/tests/client/test_stdio.py
@@ -5,6 +5,7 @@
 import tempfile
 import textwrap
 import time
+from unittest.mock import MagicMock, patch
 
 import anyio
 import pytest
@@ -648,13 +649,13 @@ async def test_stderr_captured_in_process(self):
         """Test that stderr output from a subprocess can be captured."""
         # Create a script that writes to stderr
         script = textwrap.dedent(
-            '''
+            """
             import sys
             sys.stderr.write("test error message\\n")
             sys.stderr.flush()
             # Exit immediately
             sys.exit(0)
-            '''
+            """
         )
 
         server_params = StdioServerParameters(
@@ -674,7 +675,7 @@ async def test_stderr_with_continuous_output(self):
         """Test that continuous stderr output doesn't block the client."""
         # Create a script that writes to stderr continuously then exits
         script = textwrap.dedent(
-            '''
+            """
             import sys
             import time
 
@@ -685,7 +686,7 @@ async def test_stderr_with_continuous_output(self):
 
             # Exit after writing
             sys.exit(0)
-            '''
+            """
         )
 
         server_params = StdioServerParameters(
@@ -699,3 +700,96 @@ async def test_stderr_with_continuous_output(self):
                 await anyio.sleep(1.0)  # Wait for stderr output
 
         assert not cancel_scope.cancelled_caught, "stdio_client should handle continuous stderr output"
+
+    def test_jupyter_detection_ipkernel_app(self):
+        """Test that _is_jupyter_environment returns True for IPKernelApp config."""
+        mock_ipython_instance = MagicMock()
+        mock_ipython_instance.config = {"IPKernelApp": {}}
+        mock_ipython_instance.__class__ = type("TerminalInteractiveShell", (), {})
+
+        mock_ipython_module = MagicMock()
+        mock_ipython_module.get_ipython = MagicMock(return_value=mock_ipython_instance)
+
+        with patch.dict("sys.modules", {"IPython": mock_ipython_module}):
+            result = _is_jupyter_environment()
+            assert result is True
+
+    def test_jupyter_detection_zmq_shell(self):
+        """Test that _is_jupyter_environment returns True for ZMQInteractiveShell."""
+        mock_ipython_instance = MagicMock()
+        mock_ipython_instance.config = {}
+        mock_ipython_instance.__class__ = type("ZMQInteractiveShell", (), {})
+
+        mock_ipython_module = MagicMock()
+        mock_ipython_module.get_ipython = MagicMock(return_value=mock_ipython_instance)
+
+        with patch.dict("sys.modules", {"IPython": mock_ipython_module}):
+            result = _is_jupyter_environment()
+            assert result is True
+
+    def test_jupyter_detection_non_notebook_ipython(self):
+        """Test that _is_jupyter_environment returns False for plain IPython terminal."""
+        mock_ipython_instance = MagicMock()
+        mock_ipython_instance.config = {}
+        mock_ipython_instance.__class__ = type("TerminalInteractiveShell", (), {})
+
+        mock_ipython_module = MagicMock()
+        mock_ipython_module.get_ipython = MagicMock(return_value=mock_ipython_instance)
+
+        with patch.dict("sys.modules", {"IPython": mock_ipython_module}):
+            result = _is_jupyter_environment()
+            assert result is False
+
+    @pytest.mark.anyio
+    async def test_stderr_reader_jupyter_mode(self):
+        """Test that stderr is captured and printed in Jupyter mode."""
+        script = textwrap.dedent(
+            """
+            import sys
+            sys.stderr.write("jupyter error output\\n")
+            sys.stderr.flush()
+            sys.exit(0)
+            """
+        )
+
+        server_params = StdioServerParameters(
+            command=sys.executable,
+            args=["-c", script],
+        )
+
+        # Mock _is_jupyter_environment to return True to exercise stderr_reader
+        with patch("mcp.client.stdio._is_jupyter_environment", return_value=True):
+            with anyio.move_on_after(5.0) as cancel_scope:
+                async with stdio_client(server_params) as (read_stream, write_stream):
+                    await anyio.sleep(1.0)
+
+        assert not cancel_scope.cancelled_caught, "stdio_client should not hang in Jupyter mode"
+
+    @pytest.mark.anyio
+    async def test_stderr_reader_jupyter_mode_continuous(self):
+        """Test that continuous stderr output is handled in Jupyter mode."""
+        script = textwrap.dedent(
+            """
+            import sys
+            import time
+
+            for i in range(3):
+                sys.stderr.write(f"jupyter stderr line {i}\\n")
+                sys.stderr.flush()
+                time.sleep(0.05)
+
+            sys.exit(0)
+            """
+        )
+
+        server_params = StdioServerParameters(
+            command=sys.executable,
+            args=["-c", script],
+        )
+
+        with patch("mcp.client.stdio._is_jupyter_environment", return_value=True):
+            with anyio.move_on_after(5.0) as cancel_scope:
+                async with stdio_client(server_params) as (read_stream, write_stream):
+                    await anyio.sleep(1.0)
+
+        assert not cancel_scope.cancelled_caught, "stdio_client should handle continuous Jupyter stderr"

From bad0a6ed1d221a0d20396707d1c2befc46470b6a Mon Sep 17 00:00:00 2001
From: BabyChrist666 <istiaqrafi39@gmail.com>
Date: Sun, 15 Feb 2026 00:53:15 -0500
Subject: [PATCH 3/7] Fix remaining pre-commit issues and add missing branch
 coverage

- Remove unused Union import in win32/utilities.py, use | syntax
- Add test for get_ipython() returning None (covers 43->52 branch)

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
---
 src/mcp/os/win32/utilities.py | 6 +++---
 tests/client/test_stdio.py    | 9 +++++++++
 2 files changed, 12 insertions(+), 3 deletions(-)

diff --git a/src/mcp/os/win32/utilities.py b/src/mcp/os/win32/utilities.py
index 5be39864f..f1eae56ec 100644
--- a/src/mcp/os/win32/utilities.py
+++ b/src/mcp/os/win32/utilities.py
@@ -5,7 +5,7 @@
 import subprocess
 import sys
 from pathlib import Path
-from typing import BinaryIO, TextIO, Union, cast
+from typing import BinaryIO, TextIO, cast
 
 import anyio
 from anyio import to_thread
@@ -137,7 +137,7 @@ async def create_windows_process(
     command: str,
     args: list[str],
     env: dict[str, str] | None = None,
-    errlog: Union[TextIO, int, 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.
@@ -196,7 +196,7 @@ async def _create_windows_fallback_process(
     command: str,
     args: list[str],
     env: dict[str, str] | None = None,
-    errlog: Union[TextIO, int, None] = sys.stderr,
+    errlog: TextIO | int | None = sys.stderr,
     cwd: Path | str | None = None,
 ) -> FallbackProcess:
     """Create a subprocess using subprocess.Popen as a fallback when anyio fails.
diff --git a/tests/client/test_stdio.py b/tests/client/test_stdio.py
index 4b80c7476..7e375a410 100644
--- a/tests/client/test_stdio.py
+++ b/tests/client/test_stdio.py
@@ -740,6 +740,15 @@ def test_jupyter_detection_non_notebook_ipython(self):
             result = _is_jupyter_environment()
             assert result is False
 
+    def test_jupyter_detection_ipython_returns_none(self):
+        """Test that _is_jupyter_environment returns False when get_ipython() returns None."""
+        mock_ipython_module = MagicMock()
+        mock_ipython_module.get_ipython = MagicMock(return_value=None)
+
+        with patch.dict("sys.modules", {"IPython": mock_ipython_module}):
+            result = _is_jupyter_environment()
+            assert result is False
+
     @pytest.mark.anyio
     async def test_stderr_reader_jupyter_mode(self):
         """Test that stderr is captured and printed in Jupyter mode."""

From 4b199bc37587e55ef82072d47c464fc030337219 Mon Sep 17 00:00:00 2001
From: BabyChrist666 <istiaqrafi39@gmail.com>
Date: Sun, 15 Feb 2026 00:57:38 -0500
Subject: [PATCH 4/7] Fix pyright type errors and branch coverage in tests

- Add type: ignore comments for IPython imports (reportMissingImports, etc.)
- Add pragma: no branch for async context manager branches in Jupyter tests

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
---
 src/mcp/client/stdio.py    |  8 ++++----
 tests/client/test_stdio.py | 10 ++++++++--
 2 files changed, 12 insertions(+), 6 deletions(-)

diff --git a/src/mcp/client/stdio.py b/src/mcp/client/stdio.py
index 3b5c45d0e..b5a2d17bc 100644
--- a/src/mcp/client/stdio.py
+++ b/src/mcp/client/stdio.py
@@ -37,15 +37,15 @@ def _is_jupyter_environment() -> bool:
     """
     try:
         # Check for IPython kernel
-        from IPython import get_ipython
+        from IPython import get_ipython  # type: ignore[reportMissingImports]
 
-        ipython = get_ipython()
+        ipython = get_ipython()  # type: ignore[reportUnknownVariableType]
         if ipython is not None:
             # Check if it's a notebook kernel (not just IPython terminal)
-            if "IPKernelApp" in ipython.config:
+            if "IPKernelApp" in ipython.config:  # type: ignore[reportUnknownMemberType]
                 return True
             # Also check for ZMQInteractiveShell which indicates notebook
-            if ipython.__class__.__name__ == "ZMQInteractiveShell":
+            if ipython.__class__.__name__ == "ZMQInteractiveShell":  # type: ignore[reportUnknownMemberType]
                 return True
     except (ImportError, AttributeError):
         pass
diff --git a/tests/client/test_stdio.py b/tests/client/test_stdio.py
index 7e375a410..74342e9ec 100644
--- a/tests/client/test_stdio.py
+++ b/tests/client/test_stdio.py
@@ -769,7 +769,10 @@ async def test_stderr_reader_jupyter_mode(self):
         # Mock _is_jupyter_environment to return True to exercise stderr_reader
         with patch("mcp.client.stdio._is_jupyter_environment", return_value=True):
             with anyio.move_on_after(5.0) as cancel_scope:
-                async with stdio_client(server_params) as (read_stream, write_stream):
+                async with stdio_client(server_params) as (  # pragma: no branch
+                    read_stream,
+                    write_stream,
+                ):
                     await anyio.sleep(1.0)
 
         assert not cancel_scope.cancelled_caught, "stdio_client should not hang in Jupyter mode"
@@ -798,7 +801,10 @@ async def test_stderr_reader_jupyter_mode_continuous(self):
 
         with patch("mcp.client.stdio._is_jupyter_environment", return_value=True):
             with anyio.move_on_after(5.0) as cancel_scope:
-                async with stdio_client(server_params) as (read_stream, write_stream):
+                async with stdio_client(server_params) as (  # pragma: no branch
+                    read_stream,
+                    write_stream,
+                ):
                     await anyio.sleep(1.0)
 
         assert not cancel_scope.cancelled_caught, "stdio_client should handle continuous Jupyter stderr"

From 39a940bf94c67d2333f62763af873d8a77a72398 Mon Sep 17 00:00:00 2001
From: BabyChrist666 <istiaqrafi39@gmail.com>
Date: Sun, 15 Feb 2026 23:59:25 -0500
Subject: [PATCH 5/7] Fix pyright errors: unused variables and __class__
 assignment

- Prefix unused stream variables with _ to satisfy reportUnusedVariable
- Use real classes instead of MagicMock __class__ assignment for IPython mocks

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
---
 tests/client/test_stdio.py | 36 ++++++++++++++++++------------------
 1 file changed, 18 insertions(+), 18 deletions(-)

diff --git a/tests/client/test_stdio.py b/tests/client/test_stdio.py
index 74342e9ec..f3a1c008a 100644
--- a/tests/client/test_stdio.py
+++ b/tests/client/test_stdio.py
@@ -665,7 +665,7 @@ async def test_stderr_captured_in_process(self):
 
         # The stdio_client should handle this without hanging
         with anyio.move_on_after(3.0) as cancel_scope:
-            async with stdio_client(server_params) as (read_stream, write_stream):
+            async with stdio_client(server_params) as (_read_stream, _write_stream):
                 await anyio.sleep(0.5)  # Give process time to write and exit
 
         assert not cancel_scope.cancelled_caught, "stdio_client should not hang on stderr output"
@@ -696,19 +696,19 @@ async def test_stderr_with_continuous_output(self):
 
         # The client should handle continuous stderr without blocking
         with anyio.move_on_after(5.0) as cancel_scope:
-            async with stdio_client(server_params) as (read_stream, write_stream):
+            async with stdio_client(server_params) as (_read_stream, _write_stream):
                 await anyio.sleep(1.0)  # Wait for stderr output
 
         assert not cancel_scope.cancelled_caught, "stdio_client should handle continuous stderr output"
 
     def test_jupyter_detection_ipkernel_app(self):
         """Test that _is_jupyter_environment returns True for IPKernelApp config."""
-        mock_ipython_instance = MagicMock()
-        mock_ipython_instance.config = {"IPKernelApp": {}}
-        mock_ipython_instance.__class__ = type("TerminalInteractiveShell", (), {})
+
+        class FakeIPython:
+            config = {"IPKernelApp": {}}
 
         mock_ipython_module = MagicMock()
-        mock_ipython_module.get_ipython = MagicMock(return_value=mock_ipython_instance)
+        mock_ipython_module.get_ipython = MagicMock(return_value=FakeIPython())
 
         with patch.dict("sys.modules", {"IPython": mock_ipython_module}):
             result = _is_jupyter_environment()
@@ -716,12 +716,12 @@ def test_jupyter_detection_ipkernel_app(self):
 
     def test_jupyter_detection_zmq_shell(self):
         """Test that _is_jupyter_environment returns True for ZMQInteractiveShell."""
-        mock_ipython_instance = MagicMock()
-        mock_ipython_instance.config = {}
-        mock_ipython_instance.__class__ = type("ZMQInteractiveShell", (), {})
+
+        class ZMQInteractiveShell:
+            config: dict[str, object] = {}
 
         mock_ipython_module = MagicMock()
-        mock_ipython_module.get_ipython = MagicMock(return_value=mock_ipython_instance)
+        mock_ipython_module.get_ipython = MagicMock(return_value=ZMQInteractiveShell())
 
         with patch.dict("sys.modules", {"IPython": mock_ipython_module}):
             result = _is_jupyter_environment()
@@ -729,12 +729,12 @@ def test_jupyter_detection_zmq_shell(self):
 
     def test_jupyter_detection_non_notebook_ipython(self):
         """Test that _is_jupyter_environment returns False for plain IPython terminal."""
-        mock_ipython_instance = MagicMock()
-        mock_ipython_instance.config = {}
-        mock_ipython_instance.__class__ = type("TerminalInteractiveShell", (), {})
+
+        class TerminalInteractiveShell:
+            config: dict[str, object] = {}
 
         mock_ipython_module = MagicMock()
-        mock_ipython_module.get_ipython = MagicMock(return_value=mock_ipython_instance)
+        mock_ipython_module.get_ipython = MagicMock(return_value=TerminalInteractiveShell())
 
         with patch.dict("sys.modules", {"IPython": mock_ipython_module}):
             result = _is_jupyter_environment()
@@ -770,8 +770,8 @@ async def test_stderr_reader_jupyter_mode(self):
         with patch("mcp.client.stdio._is_jupyter_environment", return_value=True):
             with anyio.move_on_after(5.0) as cancel_scope:
                 async with stdio_client(server_params) as (  # pragma: no branch
-                    read_stream,
-                    write_stream,
+                    _read_stream,
+                    _write_stream,
                 ):
                     await anyio.sleep(1.0)
 
@@ -802,8 +802,8 @@ async def test_stderr_reader_jupyter_mode_continuous(self):
         with patch("mcp.client.stdio._is_jupyter_environment", return_value=True):
             with anyio.move_on_after(5.0) as cancel_scope:
                 async with stdio_client(server_params) as (  # pragma: no branch
-                    read_stream,
-                    write_stream,
+                    _read_stream,
+                    _write_stream,
                 ):
                     await anyio.sleep(1.0)
 

From f568ea144db9133601a37850878b3c3df7ddaba0 Mon Sep 17 00:00:00 2001
From: BabyChrist666 <istiaqrafi39@gmail.com>
Date: Mon, 16 Feb 2026 00:02:35 -0500
Subject: [PATCH 6/7] Fix pyright partially unknown type error for mock config

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
---
 tests/client/test_stdio.py | 6 ++++--
 1 file changed, 4 insertions(+), 2 deletions(-)

diff --git a/tests/client/test_stdio.py b/tests/client/test_stdio.py
index f3a1c008a..81645b699 100644
--- a/tests/client/test_stdio.py
+++ b/tests/client/test_stdio.py
@@ -718,7 +718,8 @@ def test_jupyter_detection_zmq_shell(self):
         """Test that _is_jupyter_environment returns True for ZMQInteractiveShell."""
 
         class ZMQInteractiveShell:
-            config: dict[str, object] = {}
+            def __init__(self) -> None:
+                self.config: dict[str, str] = {}
 
         mock_ipython_module = MagicMock()
         mock_ipython_module.get_ipython = MagicMock(return_value=ZMQInteractiveShell())
@@ -731,7 +732,8 @@ def test_jupyter_detection_non_notebook_ipython(self):
         """Test that _is_jupyter_environment returns False for plain IPython terminal."""
 
         class TerminalInteractiveShell:
-            config: dict[str, object] = {}
+            def __init__(self) -> None:
+                self.config: dict[str, str] = {}
 
         mock_ipython_module = MagicMock()
         mock_ipython_module.get_ipython = MagicMock(return_value=TerminalInteractiveShell())

From f12a9d37314bffb5a725d99cfd52f01297565976 Mon Sep 17 00:00:00 2001
From: BabyChrist666 <istiaqrafi39@gmail.com>
Date: Mon, 16 Feb 2026 00:05:29 -0500
Subject: [PATCH 7/7] Fix pyright: annotate FakeIPython.config type

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
---
 tests/client/test_stdio.py | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/tests/client/test_stdio.py b/tests/client/test_stdio.py
index 81645b699..2b4f833aa 100644
--- a/tests/client/test_stdio.py
+++ b/tests/client/test_stdio.py
@@ -705,7 +705,7 @@ def test_jupyter_detection_ipkernel_app(self):
         """Test that _is_jupyter_environment returns True for IPKernelApp config."""
 
         class FakeIPython:
-            config = {"IPKernelApp": {}}
+            config: dict[str, dict[str, str]] = {"IPKernelApp": {}}
 
         mock_ipython_module = MagicMock()
         mock_ipython_module.get_ipython = MagicMock(return_value=FakeIPython())