Implement cross-platform subprocess and children termination

- Add process group support on POSIX systems via start_new_session=True
- Implement _terminate_process_with_children() and _kill_process_and_children()
  functions that properly handle child process cleanup
- On POSIX: Kill entire process group at once to avoid orphaning children
  when parent dies first
- On Windows: Use taskkill /T for process tree termination
- Add pid property to FallbackProcess for consistent interface
- Add proper type annotations (Process | FallbackProcess)
- Replace existing termination logic with new unified approach

This addresses issues from PRs #850 (npx child processes) and #729
(Windows process trees) in a unified, dependency-free way.

Reported-by:surya-prakash-susarla
Reported-by:jingx8885
Github-Issue:#547

Add regression tests for child process termination

- test_stdio_client_child_process_cleanup: Verifies that child processes
  spawned by the parent (like npx spawning node) are properly terminated
- test_stdio_client_nested_process_tree: Tests that deeply nested process
  trees (parent -> child -> grandchild) are all terminated
- test_stdio_client_early_parent_exit: Tests the race condition where
  parent exits during cleanup but children are still terminated via
  process group

These tests verify the fix works correctly across different subprocess
spawning scenarios without requiring external dependencies like psutil.

Author: felixweinberger

src/mcp/client/stdio/__init__.py

@@ -1,11 +1,14 @@
 import os
+import signal
+import subprocess
 import sys
 from contextlib import asynccontextmanager
 from pathlib import Path
 from typing import Literal, TextIO
 
 import anyio
 import anyio.lowlevel
+from anyio.abc import Process
 from anyio.streams.memory import MemoryObjectReceiveStream, MemoryObjectSendStream
 from anyio.streams.text import TextReceiveStream
 from pydantic import BaseModel, Field
@@ -14,6 +17,7 @@
 from mcp.shared.message import SessionMessage
 
 from .win32 import (
+    FallbackProcess,
     create_windows_process,
     get_windows_executable_command,
 )
@@ -193,18 +197,9 @@ async def stdin_writer():
                 with anyio.fail_after(2.0):
                     await process.wait()
             except TimeoutError:
-                # Process didn't exit from stdin closure, escalate to SIGTERM
-                try:
-                    process.terminate()
-                    with anyio.fail_after(2.0):
-                        await process.wait()
-                except TimeoutError:
-                    # Process didn't respond to SIGTERM, force kill it
-                    process.kill()
-                    await process.wait()
-                except ProcessLookupError:
-                    # Process already exited, which is fine
-                    pass
+                # Process didn't exit from stdin closure, use our termination function
+                # that handles child processes properly
+                await _terminate_process_with_children(process)
             except ProcessLookupError:
                 # Process already exited, which is fine
                 pass
@@ -245,6 +240,79 @@ async def _create_platform_compatible_process(
     if sys.platform == "win32":
         process = await create_windows_process(command, args, env, errlog, cwd)
     else:
-        process = await anyio.open_process([command, *args], env=env, stderr=errlog, cwd=cwd)
+        # POSIX: Create new process group for easier cleanup
+        process = await anyio.open_process(
+            [command, *args],
+            env=env,
+            stderr=errlog,
+            cwd=cwd,
+            start_new_session=True,  # Creates new process group
+        )
 
     return process
+
+
+async def _terminate_process_with_children(process: Process | FallbackProcess, timeout: float = 2.0) -> None:
+    """
+    Terminate a process and all its children across platforms.
+
+    There's no cross-platform way in the stdlib to kill a process AND its children,
+    so we need platform-specific handling:
+    - POSIX: Use process groups and os.killpg()
+    - Windows: Use taskkill with /T flag for tree termination
+
+    Args:
+        process: The process to terminate
+        timeout: Time to wait for graceful termination before force killing
+    """
+    if sys.platform != "win32":
+        # POSIX: Kill entire process group to avoid orphaning children
+        pid = getattr(process, "pid", None)
+        if not pid:
+            return
+
+        try:
+            # Try graceful termination of the group first
+            pgid = os.getpgid(pid)
+            os.killpg(pgid, signal.SIGTERM)
+
+            # Wait for termination
+            with anyio.fail_after(timeout):
+                await process.wait()
+        except TimeoutError:
+            # Force kill the process group
+            try:
+                pgid = os.getpgid(pid)
+                os.killpg(pgid, signal.SIGKILL)
+            except (OSError, ProcessLookupError):
+                pass
+        except (OSError, ProcessLookupError):
+            # Process or group already dead
+            pass
+    else:
+        # Windows: Extract PID from FallbackProcess if needed
+        pid = getattr(process, "pid", None)
+        if pid is None:
+            popen = getattr(process, "popen", None)
+            if popen:
+                pid = getattr(popen, "pid", None)
+
+        if not pid:
+            return
+
+        # Try graceful termination first
+        try:
+            process.terminate()
+            with anyio.fail_after(timeout):
+                await process.wait()
+        except TimeoutError:
+            # Force kill using taskkill for tree termination
+            await anyio.to_thread.run_sync(
+                subprocess.run,
+                ["taskkill", "/F", "/T", "/PID", str(pid)],
+                capture_output=True,
+                shell=False,
+                check=False,
+            )
+        except ProcessLookupError:
+            pass

src/mcp/client/stdio/win32.py

@@ -101,6 +101,11 @@ def kill(self) -> None:
         """Kill the subprocess immediately (alias for terminate)."""
         self.terminate()
 
+    @property
+    def pid(self) -> int:
+        """Return the process ID."""
+        return self.popen.pid
+
 
 # ------------------------
 # Updated function