Implement MCP spec-compliant stdio shutdown sequence

Following the MCP spec recommendations, the shutdown sequence now:
1. Closes stdin first to allow graceful server exit
2. Waits 2 seconds for the server to exit on its own
3. Only escalates to SIGTERM if the server doesn't exit
4. Finally uses SIGKILL as a last resort

This unified approach works consistently across all platforms and gives
well-behaved servers a chance to exit cleanly without signals.

Co-Authored-By: davenpi <[email protected]>

Add tests for MCP spec-compliant stdio shutdown

Added two tests to validate the stdin-first shutdown behavior:

1. test_stdio_client_graceful_stdin_exit: Verifies that a well-behaved
   server exits cleanly when stdin is closed, without needing signals

2. test_stdio_client_stdin_close_ignored: Tests proper escalation to
   SIGTERM when a process ignores stdin closure

These tests ensure the MCP spec shutdown sequence works correctly and
provides graceful exit opportunities before using forceful termination.

Author: felixweinberger

src/mcp/client/stdio/__init__.py

@@ -177,17 +177,38 @@ async def stdin_writer():
         try:
             yield read_stream, write_stream
         finally:
-            # Clean up process to prevent any dangling orphaned processes
+            # MCP spec: stdio shutdown sequence
+            # 1. Close input stream to server
+            # 2. Wait for server to exit, or send SIGTERM if it doesn't exit in time
+            # 3. Send SIGKILL if still not exited
+            if process.stdin:
+                try:
+                    await process.stdin.aclose()
+                except Exception:
+                    # stdin might already be closed, which is fine
+                    pass
+
             try:
-                process.terminate()
+                # Give the process time to exit gracefully after stdin closes
                 with anyio.fail_after(2.0):
                     await process.wait()
             except TimeoutError:
-                # If process doesn't terminate in time, force kill it
-                process.kill()
+                # 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
             except ProcessLookupError:
                 # Process already exited, which is fine
                 pass
+
             await read_stream.aclose()
             await write_stream.aclose()
             await read_stream_writer.aclose()

src/mcp/client/stdio/win32.py

@@ -8,7 +8,7 @@
 from pathlib import Path
 from typing import BinaryIO, TextIO, cast
 
-from anyio import to_thread
+import anyio
 from anyio.streams.file import FileReadStream, FileWriteStream
 
 
@@ -73,9 +73,20 @@ async def __aexit__(
         exc_val: BaseException | None,
         exc_tb: object | None,
     ) -> None:
-        """Terminate and wait on process exit inside a thread."""
-        self.popen.terminate()
-        await to_thread.run_sync(self.popen.wait)
+        """Clean up process and streams.
+
+        Attempts to terminate the process, but doesn't fail if termination
+        is not possible (e.g., process already dead or being handled elsewhere).
+        """
+        # Try to terminate the process, but don't fail if it's already being handled
+        try:
+            self.popen.terminate()
+            # Use our non-blocking wait with a short timeout
+            with anyio.move_on_after(0.5):
+                await self.wait()
+        except (ProcessLookupError, OSError):
+            # Process already dead or being handled elsewhere
+            pass
 
         # Close the file handles to prevent ResourceWarning
         if self.stdin:
@@ -91,7 +102,11 @@ async def __aexit__(
 
     async def wait(self):
         """Async wait for process completion."""
-        return await to_thread.run_sync(self.popen.wait)
+        # Poll the process status instead of blocking wait
+        # This allows anyio timeouts to work properly
+        while self.popen.poll() is None:
+            await anyio.sleep(0.1)
+        return self.popen.returncode
 
     def terminate(self):
         """Terminate the subprocess immediately."""

tests/client/test_stdio.py

@@ -120,12 +120,13 @@ async def test_stdio_client_universal_cleanup():
     )
 
     server_params = StdioServerParameters(
-        command=sys.executable,
+        command="python",
         args=["-c", long_running_script],
     )
 
     start_time = time.time()
 
+    # Use move_on_after which is more reliable for cleanup scenarios
     with anyio.move_on_after(8.0) as cancel_scope:
         async with stdio_client(server_params) as (read_stream, write_stream):
             # Immediately exit - this triggers cleanup while process is still running
@@ -134,16 +135,16 @@ async def test_stdio_client_universal_cleanup():
         end_time = time.time()
         elapsed = end_time - start_time
 
-        # On Windows: 2s (stdin wait) + 2s (terminate wait) + overhead = ~5s expected
-        assert elapsed < 6.0, (
-            f"stdio_client cleanup took {elapsed:.1f} seconds, expected < 6.0 seconds. "
+        # Key assertion: Should complete quickly due to timeout mechanism
+        assert elapsed < 5.0, (
+            f"stdio_client cleanup took {elapsed:.1f} seconds, expected < 5.0 seconds. "
             f"This suggests the timeout mechanism may not be working properly."
         )
 
     # Check if we timed out
     if cancel_scope.cancelled_caught:
         pytest.fail(
-            "stdio_client cleanup timed out after 8.0 seconds. "
+            "stdio_client cleanup timed out after 6.0 seconds. "
             "This indicates the cleanup mechanism is hanging and needs fixing."
         )
 
@@ -212,3 +213,119 @@ def sigint_handler(signum, frame):
             )
         else:
             raise
+
+
+@pytest.mark.anyio
+async def test_stdio_client_graceful_stdin_exit():
+    """
+    Test that a process exits gracefully when stdin is closed,
+    without needing SIGTERM or SIGKILL.
+    """
+    # Create a Python script that exits when stdin is closed
+    script_content = textwrap.dedent(
+        """
+        import sys
+        
+        # Read from stdin until it's closed
+        try:
+            while True:
+                line = sys.stdin.readline()
+                if not line:  # EOF/stdin closed
+                    break
+        except:
+            pass
+        
+        # Exit gracefully
+        sys.exit(0)
+        """
+    )
+
+    server_params = StdioServerParameters(
+        command=sys.executable,
+        args=["-c", script_content],
+    )
+
+    start_time = time.time()
+
+    # Use anyio timeout to prevent test from hanging forever
+    with anyio.move_on_after(5.0) as cancel_scope:
+        async with stdio_client(server_params) as (read_stream, write_stream):
+            # Let the process start and begin reading stdin
+            await anyio.sleep(0.2)
+            # Exit context triggers cleanup - process should exit from stdin closure
+            pass
+
+    if cancel_scope.cancelled_caught:
+        pytest.fail(
+            "stdio_client cleanup timed out after 5.0 seconds. "
+            "Process should have exited gracefully when stdin was closed."
+        )
+
+    end_time = time.time()
+    elapsed = end_time - start_time
+
+    # Should complete quickly with just stdin closure (no signals needed)
+    assert elapsed < 3.0, (
+        f"stdio_client cleanup took {elapsed:.1f} seconds for stdin-aware process. "
+        f"Expected < 3.0 seconds since process should exit on stdin closure."
+    )
+
+
+@pytest.mark.anyio
+async def test_stdio_client_stdin_close_ignored():
+    """
+    Test that when a process ignores stdin closure, the shutdown sequence
+    properly escalates to SIGTERM.
+    """
+    # Create a Python script that ignores stdin closure but responds to SIGTERM
+    script_content = textwrap.dedent(
+        """
+        import signal
+        import sys
+        import time
+        
+        # Set up SIGTERM handler to exit cleanly
+        def sigterm_handler(signum, frame):
+            sys.exit(0)
+        
+        signal.signal(signal.SIGTERM, sigterm_handler)
+        
+        # Close stdin immediately to simulate ignoring it
+        sys.stdin.close()
+        
+        # Keep running until SIGTERM
+        while True:
+            time.sleep(0.1)
+        """
+    )
+
+    server_params = StdioServerParameters(
+        command=sys.executable,
+        args=["-c", script_content],
+    )
+
+    start_time = time.time()
+
+    # Use anyio timeout to prevent test from hanging forever
+    with anyio.move_on_after(7.0) as cancel_scope:
+        async with stdio_client(server_params) as (read_stream, write_stream):
+            # Let the process start
+            await anyio.sleep(0.2)
+            # Exit context triggers cleanup
+            pass
+
+    if cancel_scope.cancelled_caught:
+        pytest.fail(
+            "stdio_client cleanup timed out after 7.0 seconds. "
+            "Process should have been terminated via SIGTERM escalation."
+        )
+
+    end_time = time.time()
+    elapsed = end_time - start_time
+
+    # Should take ~2 seconds (stdin close timeout) before SIGTERM is sent
+    # Total time should be between 2-4 seconds
+    assert 1.5 < elapsed < 4.5, (
+        f"stdio_client cleanup took {elapsed:.1f} seconds for stdin-ignoring process. "
+        f"Expected between 2-4 seconds (2s stdin timeout + termination time)."
+    )