Fix child process cleanup in stdio termination

When terminating MCP servers, child processes were being orphaned
because only the parent process was killed. This caused resource
leaks and prevented proper cleanup, especially with tools like npx
that spawn child processes for the actual server implementation.

This was happening on both POSIX and Windows systems - however
because of implementation details, resolving this is non-trivial
and requires introducing psutil to introduce cross-platform
utilities for dealing with children and process trees.

This addresses critical issues where MCP servers using process
spawning tools would leave zombie processes running after client
shutdown.

resolves #850
resolves #729

Co-authored-by: jingx8885 <jingxu8885@qq.com>
Co-authored-by: Surya Prakash Susarla <susarla.surya.prakash.1998@gmail.com>

Author: 3 people

pyproject.toml

@@ -32,6 +32,7 @@ dependencies = [
     "pydantic-settings>=2.5.2",
     "uvicorn>=0.23.1; sys_platform != 'emscripten'",
     "jsonschema>=4.20.0",
+    "pywin32>=310; sys_platform == 'win32'",
 ]
 
 [project.optional-dependencies]
@@ -124,5 +125,7 @@ filterwarnings = [
     # This should be fixed on Uvicorn's side.
     "ignore::DeprecationWarning:websockets",
     "ignore:websockets.server.WebSocketServerProtocol is deprecated:DeprecationWarning",
-    "ignore:Returning str or bytes.*:DeprecationWarning:mcp.server.lowlevel"
+    "ignore:Returning str or bytes.*:DeprecationWarning:mcp.server.lowlevel",
+    # pywin32 internal deprecation warning
+    "ignore:getargs.*The 'u' format is deprecated:DeprecationWarning"
 ]

src/mcp/client/stdio/__init__.py

@@ -1,11 +1,13 @@
 import os
+import signal
 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,8 +16,10 @@
 from mcp.shared.message import SessionMessage
 
 from .win32 import (
+    FallbackProcess,
     create_windows_process,
     get_windows_executable_command,
+    terminate_windows_process_tree,
 )
 
 # Environment variables to inherit by default
@@ -184,7 +188,7 @@ async def stdin_writer():
                     await process.wait()
             except TimeoutError:
                 # If process doesn't terminate in time, force kill it
-                process.kill()
+                await _terminate_process_tree(process)
             except ProcessLookupError:
                 # Process already exited, which is fine
                 pass
@@ -219,11 +223,80 @@ async def _create_platform_compatible_process(
 ):
     """
     Creates a subprocess in a platform-compatible way.
-    Returns a process handle.
+
+    Unix: Creates process in a new session/process group for killpg support
+    Windows: Creates process in a Job Object for reliable child termination
     """
     if sys.platform == "win32":
+        # Windows: Use Job Objects for proper process tree management
         process = await create_windows_process(command, args, env, errlog, cwd)
     else:
-        process = await anyio.open_process([command, *args], env=env, stderr=errlog, cwd=cwd)
+        # Unix: Create process in new session for process group termination
+        process = await anyio.open_process(
+            [command, *args],
+            env=env,
+            stderr=errlog,
+            cwd=cwd,
+            start_new_session=True,
+        )
 
     return process
+
+
+async def _terminate_process_tree(process: Process | FallbackProcess, timeout: float = 2.0) -> None:
+    """
+    Terminate a process and all its children using platform-specific methods.
+
+    Unix: Uses os.killpg() for atomic process group termination
+    Windows: Uses Job Objects via pywin32 for reliable child process cleanup
+    """
+    if sys.platform == "win32":
+        # Windows: Use Job Object termination
+        await terminate_windows_process_tree(process)
+    else:
+        # Unix: Use process groups for atomic termination
+        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:
+            # Get process group ID (we use start_new_session=True)
+            pgid = os.getpgid(pid)
+
+            # Send SIGTERM to entire process group atomically
+            os.killpg(pgid, signal.SIGTERM)
+
+            # Wait for graceful termination
+            deadline = anyio.current_time() + timeout
+            while anyio.current_time() < deadline:
+                try:
+                    # Check if process group still exists (signal 0 = check only)
+                    os.killpg(pgid, 0)
+                    await anyio.sleep(0.1)
+                except ProcessLookupError:
+                    # Process group terminated successfully
+                    return
+
+            # Force kill if still alive after timeout
+            try:
+                os.killpg(pgid, signal.SIGKILL)
+            except ProcessLookupError:
+                # Already dead
+                pass
+
+        except (ProcessLookupError, PermissionError, OSError):
+            # Fall back to simple terminate if process group approach fails
+            try:
+                process.terminate()
+                with anyio.fail_after(timeout):
+                    await process.wait()
+            except Exception:
+                try:
+                    process.kill()
+                except Exception:
+                    pass

src/mcp/client/stdio/win32.py

@@ -6,13 +6,28 @@
 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
 from anyio.abc import Process
 from anyio.streams.file import FileReadStream, FileWriteStream
 
+# Windows-specific imports for Job Objects
+if sys.platform == "win32":
+    import pywintypes
+    import win32api
+    import win32con
+    import win32job
+else:
+    # Type stubs for non-Windows platforms
+    win32api = None
+    win32con = None
+    win32job = None
+    pywintypes = None
+
+JobHandle = int
+
 
 def get_windows_executable_command(command: str) -> str:
     """
@@ -103,6 +118,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
@@ -117,13 +137,16 @@ async def create_windows_process(
     cwd: Path | str | None = None,
 ) -> Process | FallbackProcess:
     """
-    Creates a subprocess in a Windows-compatible way.
+    Creates a subprocess in a Windows-compatible way with Job Object support.
 
     Attempt to use anyio's open_process for async subprocess creation.
     In some cases this will throw NotImplementedError on Windows, e.g.
     when using the SelectorEventLoop which does not support async subprocesses.
     In that case, we fall back to using subprocess.Popen.
 
+    The process is automatically added to a Job Object to ensure all child
+    processes are terminated when the parent is terminated.
+
     Args:
         command (str): The executable to run
         args (list[str]): List of command line arguments
@@ -132,8 +155,12 @@ async def create_windows_process(
         cwd (Path | str | None): Working directory for the subprocess
 
     Returns:
-        FallbackProcess: Async-compatible subprocess with stdin and stdout streams
+        Process | FallbackProcess: Async-compatible subprocess with stdin and stdout streams
     """
+    # Create a job object for this process
+    job = _create_job_object()
+    process = None
+
     try:
         # First try using anyio with Windows-specific flags to hide console window
         process = await anyio.open_process(
@@ -146,10 +173,9 @@ async def create_windows_process(
             stderr=errlog,
             cwd=cwd,
         )
-        return process
     except NotImplementedError:
         # Windows often doesn't support async subprocess creation, use fallback
-        return await _create_windows_fallback_process(command, args, env, errlog, cwd)
+        process = await _create_windows_fallback_process(command, args, env, errlog, cwd)
     except Exception:
         # Try again without creation flags
         process = await anyio.open_process(
@@ -158,7 +184,9 @@ async def create_windows_process(
             stderr=errlog,
             cwd=cwd,
         )
-        return process
+
+    _maybe_assign_process_to_job(process, job)
+    return process
 
 
 async def _create_windows_fallback_process(
@@ -171,7 +199,8 @@ async def _create_windows_fallback_process(
     """
     Create a subprocess using subprocess.Popen as a fallback when anyio fails.
 
-    This function wraps the sync subprocess.Popen in an async-compatible interface.
+    This function wraps the sync subprocess.Popen in an async-compatible interface
+    and optionally assigns it to a job object.
     """
     try:
         # Try launching with creationflags to avoid opening a new console window
@@ -185,8 +214,6 @@ async def _create_windows_fallback_process(
             bufsize=0,  # Unbuffered output
             creationflags=getattr(subprocess, "CREATE_NO_WINDOW", 0),
         )
-        return FallbackProcess(popen_obj)
-
     except Exception:
         # If creationflags failed, fallback without them
         popen_obj = subprocess.Popen(
@@ -198,4 +225,108 @@ async def _create_windows_fallback_process(
             cwd=cwd,
             bufsize=0,
         )
-        return FallbackProcess(popen_obj)
+    process = FallbackProcess(popen_obj)
+    return process
+
+
+def _create_job_object() -> int | None:
+    """
+    Create a Windows Job Object configured to terminate all processes when closed.
+
+    Returns:
+        The job object handle, or None if creation failed.
+    """
+    if sys.platform != "win32" or not win32job:
+        return None
+
+    try:
+        # Create an unnamed job object
+        job = win32job.CreateJobObject(None, "")
+
+        # Query current job information
+        extended_info = win32job.QueryInformationJobObject(job, win32job.JobObjectExtendedLimitInformation)
+
+        # Set the job to terminate all processes when the handle is closed
+        extended_info["BasicLimitInformation"]["LimitFlags"] |= win32job.JOB_OBJECT_LIMIT_KILL_ON_JOB_CLOSE
+
+        # Apply the updated job information
+        win32job.SetInformationJobObject(job, win32job.JobObjectExtendedLimitInformation, extended_info)
+
+        return job
+    except Exception:
+        # If job creation fails, return None
+        return None
+
+
+def _maybe_assign_process_to_job(process: Union[Process, "FallbackProcess"], job: JobHandle | None) -> None:
+    """
+    Try to assign a process to a job object. If assignment fails
+    for any reason, the job handle is closed.
+
+    Args:
+        process: The process to assign to the job
+        job: The job object handle (may be None)
+    """
+    if not job:
+        return
+
+    if sys.platform != "win32" or not win32api or not win32con or not win32job:
+        return
+
+    try:
+        # Open the process with required permissions
+        process_handle = win32api.OpenProcess(
+            win32con.PROCESS_SET_QUOTA | win32con.PROCESS_TERMINATE, False, process.pid
+        )
+        if not process_handle:
+            raise Exception("Failed to open process handle")
+
+        try:
+            # Assign process to job
+            win32job.AssignProcessToJobObject(job, process_handle)
+            # Store job on process for later cleanup
+            process._job_object = job  # type: ignore
+        finally:
+            # Always close the process handle
+            win32api.CloseHandle(process_handle)
+    except Exception:
+        # If we can't assign to job, close it
+        if win32api:
+            win32api.CloseHandle(job)
+
+
+async def terminate_windows_process_tree(process: Union[Process, "FallbackProcess"]) -> None:
+    """
+    Terminate a process and all its children on Windows.
+
+    If the process has an associated job object, it will be terminated.
+    Otherwise, falls back to basic process termination.
+
+    Args:
+        process: The process to terminate
+    """
+    if sys.platform != "win32":
+        return
+
+    # Check if process has a job object
+    job = getattr(process, "_job_object", None)
+    if job and win32job:
+        try:
+            # Terminate all processes in the job (exit code 1)
+            win32job.TerminateJobObject(job, 1)
+        except Exception:
+            # Job might already be terminated
+            pass
+        finally:
+            if win32api:
+                try:
+                    # Close the job handle
+                    win32api.CloseHandle(job)
+                except Exception:
+                    pass
+
+    # Always try to terminate the process itself as well
+    try:
+        process.terminate()
+    except Exception:
+        pass