Simplify child process cleanup tests and remove cruft

Remove unnecessary complexity from the child process termination tests
introduced during development iterations:

- Eliminate WindowsProcessWrapper class - anyio's process handling
  works correctly on both Unix and Windows without this wrapper
- Remove escape_path_for_python() function - Python's repr() handles
  path escaping correctly across platforms
- Unify process creation code paths - use anyio.open_process()
  consistently with platform-appropriate start_new_session flag
- Remove [[tool.uv.index]] from pyproject.toml - this is the default
  PyPI index and doesn't need explicit configuration

The tests remain functionally identical but are now simpler and more
maintainable. All child process cleanup functionality continues to
work correctly on both Unix and Windows.

Author: felixweinberger

pyproject.toml

@@ -111,9 +111,6 @@ members = ["examples/servers/*"]
 [tool.uv.sources]
 mcp = { workspace = true }
 
-[[tool.uv.index]]
-url = "https://pypi.org/simple"
-
 [tool.pytest.ini_options]
 log_cli = true
 xfail_strict = true

tests/client/test_stdio.py

@@ -22,31 +22,6 @@
 python: str = shutil.which("python")  # type: ignore
 
 
-def escape_path_for_python(path: str) -> str:
-    """Escape a file path for use in Python code strings."""
-    # Use forward slashes which work on all platforms and don't need escaping
-    return repr(path.replace("\\", "/"))
-
-
-class WindowsProcessWrapper:
-    """Minimal wrapper for subprocess.Popen to work with anyio-style process interface."""
-
-    def __init__(self, popen):
-        self.pid = popen.pid
-        self._popen = popen
-        # Add popen attribute for compatibility with _terminate_process_with_children
-        self.popen = popen
-
-    async def wait(self):
-        while self._popen.poll() is None:
-            await anyio.sleep(0.1)
-        return self._popen.returncode
-
-    def terminate(self):
-        self._popen.terminate()
-
-    def kill(self):
-        self._popen.kill()
 
 
 @pytest.mark.anyio
@@ -399,13 +374,13 @@ async def test_stdio_client_child_process_cleanup():
             import os
 
             # Mark that parent started
-            with open({escape_path_for_python(parent_marker)}, 'w') as f:
+            with open({repr(parent_marker)}, 'w') as f:
                 f.write('parent started\\n')
 
             # Child script that writes continuously
             child_script = f'''
             import time
-            with open({escape_path_for_python(marker_file)}, 'a') as f:
+            with open({repr(marker_file)}, 'a') as f:
                 while True:
                     f.write(f"{time.time()}")
                     f.flush()
@@ -423,15 +398,11 @@ async def test_stdio_client_child_process_cleanup():
 
         print("\nStarting child process termination test...")
 
-        # Start the parent process directly with process group
-        if sys.platform == "win32":
-            # Windows: Don't use CREATE_NEW_PROCESS_GROUP as it isolates the process
-            # Instead, let it inherit the parent's console which allows taskkill /T to work
-            popen = subprocess.Popen([sys.executable, "-c", parent_script])
-            proc = WindowsProcessWrapper(popen)
-        else:
-            # Unix: Use start_new_session for process group creation
-            proc = await anyio.open_process([sys.executable, "-c", parent_script], start_new_session=True)
+        # Start the parent process using anyio (cross-platform)
+        proc = await anyio.open_process(
+            [sys.executable, "-c", parent_script],
+            start_new_session=(sys.platform != "win32")  # Only on Unix
+        )
 
         # Wait for processes to start
         await anyio.sleep(0.5)
@@ -509,7 +480,7 @@ async def test_stdio_client_nested_process_tree():
 
             # Grandchild just writes to file
             grandchild_script = \"\"\"import time
-            with open({escape_path_for_python(grandchild_file)}, 'a') as f:
+            with open({repr(grandchild_file)}, 'a') as f:
                 while True:
                     f.write(f"gc {{time.time()}}")
                     f.flush()
@@ -519,7 +490,7 @@ async def test_stdio_client_nested_process_tree():
             subprocess.Popen([sys.executable, '-c', grandchild_script])
 
             # Child writes to its file
-            with open({escape_path_for_python(child_file)}, 'a') as f:
+            with open({repr(child_file)}, 'a') as f:
                 while True:
                     f.write(f"c {time.time()}")
                     f.flush()
@@ -529,23 +500,19 @@ async def test_stdio_client_nested_process_tree():
             subprocess.Popen([sys.executable, '-c', child_script])
 
             # Parent writes to its file
-            with open({escape_path_for_python(parent_file)}, 'a') as f:
+            with open({repr(parent_file)}, 'a') as f:
                 while True:
                     f.write(f"p {time.time()}")
                     f.flush()
                     time.sleep(0.1)
             """
         )
 
-        # Start parent process directly
-        if sys.platform == "win32":
-            # Windows: Don't use CREATE_NEW_PROCESS_GROUP as it isolates the process
-            # Instead, let it inherit the parent's console which allows taskkill /T to work
-            popen = subprocess.Popen([sys.executable, "-c", parent_script])
-            proc = WindowsProcessWrapper(popen)
-        else:
-            # Unix: Use start_new_session for process group creation
-            proc = await anyio.open_process([sys.executable, "-c", parent_script], start_new_session=True)
+        # Start parent process using anyio (cross-platform)
+        proc = await anyio.open_process(
+            [sys.executable, "-c", parent_script],
+            start_new_session=(sys.platform != "win32")  # Only on Unix
+        )
 
         # Let all processes start
         await anyio.sleep(1.0)
@@ -607,7 +574,7 @@ async def test_stdio_client_early_parent_exit():
 
             # Child that continues running
             child_script = f'''import time
-            with open({escape_path_for_python(marker_file)}, 'a') as f:
+            with open({repr(marker_file)}, 'a') as f:
                 while True:
                     f.write(f"child {time.time()}")
                     f.flush()
@@ -628,15 +595,11 @@ def handle_term(sig, frame):
             """
         )
 
-        # Start the parent process
-        if sys.platform == "win32":
-            # Windows: Don't use CREATE_NEW_PROCESS_GROUP as it isolates the process
-            # Instead, let it inherit the parent's console which allows taskkill /T to work
-            popen = subprocess.Popen([sys.executable, "-c", parent_script])
-            proc = WindowsProcessWrapper(popen)
-        else:
-            # Unix: Use start_new_session for process group creation
-            proc = await anyio.open_process([sys.executable, "-c", parent_script], start_new_session=True)
+        # Start the parent process using anyio (cross-platform)
+        proc = await anyio.open_process(
+            [sys.executable, "-c", parent_script],
+            start_new_session=(sys.platform != "win32")  # Only on Unix
+        )
 
         # Let child start writing
         await anyio.sleep(0.5)

tests/client/test_stdio_refactored.py

@@ -0,0 +1,205 @@
+"""Refactored tests for stdio client child process cleanup."""
+import os
+import shutil
+import sys
+import tempfile
+import textwrap
+
+import pytest
+import anyio
+
+from mcp.client.stdio import _terminate_process_with_children
+
+python: str = shutil.which("python")  # type: ignore
+
+
+@pytest.mark.anyio
+async def test_stdio_client_child_process_cleanup():
+    """
+    Test that child processes are properly terminated when the parent is killed.
+    This addresses the issue where processes like npx spawn child processes
+    that need to be cleaned up.
+    """
+    # Create marker files
+    parent_marker = tempfile.mktemp(suffix=".txt")
+    child_marker = tempfile.mktemp(suffix=".txt")
+
+    try:
+        # Parent script that spawns a child process
+        parent_script = textwrap.dedent(
+            f"""
+import subprocess
+import sys
+import time
+
+# Mark that parent started
+with open({repr(parent_marker)}, 'w') as f:
+    f.write('parent started\\n')
+
+# Child script that writes continuously
+child_script = '''
+import time
+with open({repr(child_marker)}, 'a') as f:
+    while True:
+        f.write(str(time.time()) + "\\\\n")
+        f.flush()
+        time.sleep(0.1)
+'''
+
+# Start the child process
+subprocess.Popen([sys.executable, '-c', child_script])
+
+# Parent sleeps forever
+while True:
+    time.sleep(0.1)
+"""
+        ).strip()
+
+        # Start the parent process using anyio (cross-platform)
+        # Use start_new_session on Unix for process group, not needed on Windows
+        proc = await anyio.open_process(
+            [python, "-c", parent_script],
+            start_new_session=(sys.platform != "win32"),
+            stderr=None  # Let stderr go to console for debugging
+        )
+
+        # Wait for processes to start
+        await anyio.sleep(1.0)
+        
+        print(f"Parent marker path: {parent_marker}")
+        print(f"Child marker path: {child_marker}")
+        print(f"Parent marker exists: {os.path.exists(parent_marker)}")
+
+        # Verify parent started
+        assert os.path.exists(parent_marker), "Parent process didn't start"
+
+        # Verify child is writing
+        if os.path.exists(child_marker):
+            initial_size = os.path.getsize(child_marker)
+            await anyio.sleep(0.3)
+            size_after_wait = os.path.getsize(child_marker)
+            assert size_after_wait > initial_size, "Child process should be writing"
+
+        # Terminate using our function
+        await _terminate_process_with_children(proc)
+
+        # Verify processes stopped
+        await anyio.sleep(0.5)
+        if os.path.exists(child_marker):
+            size_after_cleanup = os.path.getsize(child_marker)
+            await anyio.sleep(0.5)
+            final_size = os.path.getsize(child_marker)
+            assert final_size == size_after_cleanup, f"Child still running! File grew by {final_size - size_after_cleanup} bytes"
+
+    finally:
+        # Clean up files
+        for f in [parent_marker, child_marker]:
+            if os.path.exists(f):
+                try:
+                    os.unlink(f)
+                except OSError:
+                    pass
+
+
+@pytest.mark.anyio
+async def test_stdio_client_nested_process_tree():
+    """
+    Test that a nested process tree (parent → child → grandchild) is properly cleaned up.
+    """
+    # Create marker files
+    parent_file = tempfile.mktemp(suffix=".txt")
+    child_file = tempfile.mktemp(suffix=".txt")
+    grandchild_file = tempfile.mktemp(suffix=".txt")
+
+    try:
+        # Create nested process tree
+        parent_script = textwrap.dedent(
+            f"""
+import subprocess
+import sys
+import time
+
+# Child script that spawns grandchild
+child_script = f'''
+import subprocess
+import sys
+import time
+
+# Grandchild script
+grandchild_script = \"\"\"
+import time
+with open({repr(grandchild_file)}, 'a') as f:
+    while True:
+        f.write(str(time.time()) + "\\\\n")
+        f.flush()
+        time.sleep(0.1)
+\"\"\"
+
+# Spawn grandchild
+subprocess.Popen([sys.executable, '-c', grandchild_script])
+
+# Child writes to its file
+with open({repr(child_file)}, 'a') as f:
+    while True:
+        f.write(str(time.time()) + "\\\\n")
+        f.flush()
+        time.sleep(0.1)
+'''
+
+# Spawn child process
+subprocess.Popen([sys.executable, '-c', child_script])
+
+# Parent writes to its file
+with open({repr(parent_file)}, 'a') as f:
+    while True:
+        f.write(str(time.time()) + "\\n")
+        f.flush()
+        time.sleep(0.1)
+"""
+        ).strip()
+
+        # Start parent process
+        proc = await anyio.open_process(
+            [python, "-c", parent_script],
+            start_new_session=(sys.platform != "win32")
+        )
+
+        # Let all processes start
+        await anyio.sleep(1.0)
+
+        # Verify all are writing
+        for file_path, name in [
+            (parent_file, "parent"),
+            (child_file, "child"),
+            (grandchild_file, "grandchild"),
+        ]:
+            if os.path.exists(file_path):
+                initial_size = os.path.getsize(file_path)
+                await anyio.sleep(0.3)
+                new_size = os.path.getsize(file_path)
+                assert new_size > initial_size, f"{name} process should be writing"
+
+        # Terminate all processes
+        await _terminate_process_with_children(proc)
+
+        # Verify all stopped
+        await anyio.sleep(0.5)
+        for file_path, name in [
+            (parent_file, "parent"),
+            (child_file, "child"),
+            (grandchild_file, "grandchild"),
+        ]:
+            if os.path.exists(file_path):
+                size_before = os.path.getsize(file_path)
+                await anyio.sleep(0.5)
+                size_after = os.path.getsize(file_path)
+                assert size_after == size_before, f"{name} still running! File grew by {size_after - size_before} bytes"
+
+    finally:
+        # Clean up files
+        for f in [parent_file, child_file, grandchild_file]:
+            if os.path.exists(f):
+                try:
+                    os.unlink(f)
+                except OSError:
+                    pass