feat(stdio): support preexec_fn and process_group in StdioServerParameters (#3457)

Add preexec_fn and process_group to StdioServerParameters, forwarding them through stdio_client and _create_platform_compatible_process to anyio.open_process on POSIX systems. Includes unit tests for parameter parsing, forwarding, and child-process execution.

Author: connerkup

src/mcp/client/stdio.py

@@ -10,18 +10,19 @@
 
 import logging
 import os
+import subprocess
 import sys
-from collections.abc import AsyncGenerator
+from collections.abc import AsyncGenerator, Callable
 from contextlib import asynccontextmanager, suppress
 from pathlib import Path
-from typing import Literal, TextIO
+from typing import Any, Literal, TextIO
 
 import anyio
 import anyio.lowlevel
 import mcp_types as types
 from anyio.abc import AsyncResource, Process
 from anyio.streams.text import TextReceiveStream
-from pydantic import BaseModel, Field
+from pydantic import BaseModel, ConfigDict, Field
 
 from mcp.client._transport import TransportStreams
 from mcp.os.posix.utilities import terminate_posix_process_tree
@@ -91,6 +92,8 @@ def get_default_environment() -> dict[str, str]:
 
 
 class StdioServerParameters(BaseModel):
+    model_config = ConfigDict(arbitrary_types_allowed=True)
+
     command: str
     """The executable to run to start the server."""
 
@@ -109,6 +112,12 @@ class StdioServerParameters(BaseModel):
     encoding_error_handler: Literal["strict", "ignore", "replace"] = "strict"
     """Encoding error handler; see https://docs.python.org/3/library/codecs.html#error-handlers."""
 
+    preexec_fn: Callable[[], Any] | None = None
+    """Function called in the child process just before the child is executed (POSIX only)."""
+
+    process_group: int | None = None
+    """Process group to set in the child process (POSIX only)."""
+
 
 @asynccontextmanager
 async def stdio_client(
@@ -122,12 +131,19 @@ async def stdio_client(
     """
     command = _get_executable_command(server.command)
 
+    extra_spawn_kwargs: dict[str, Any] = {}
+    if server.preexec_fn is not None:
+        extra_spawn_kwargs["preexec_fn"] = server.preexec_fn
+    if server.process_group is not None:
+        extra_spawn_kwargs["process_group"] = server.process_group
+
     process = await _create_platform_compatible_process(
         command=command,
         args=server.args,
         env=get_default_environment() | (server.env or {}),
         errlog=errlog,
         cwd=server.cwd,
+        **extra_spawn_kwargs,
     )
 
     # The spawn succeeded; no awaits until the task group is entered, or a
@@ -331,6 +347,8 @@ async def _create_platform_compatible_process(
     env: dict[str, str] | None = None,
     errlog: TextIO = sys.stderr,
     cwd: Path | str | None = None,
+    preexec_fn: Callable[[], Any] | None = None,
+    process_group: int | None = None,
 ) -> ServerProcess:
     """Spawns the server in its own kill scope.
 
@@ -339,13 +357,44 @@ async def _create_platform_compatible_process(
     if sys.platform == "win32":  # pragma: no cover
         return await create_windows_process(command, args, env, errlog, cwd)
     else:  # pragma: lax no cover
-        return await anyio.open_process(
-            [command, *args],
-            env=env,
-            stderr=errlog,
-            cwd=cwd,
-            start_new_session=True,
-        )
+        extra_kwargs: dict[str, Any] = {}
+        if preexec_fn is not None:
+            extra_kwargs["preexec_fn"] = preexec_fn
+        if process_group is not None:
+            extra_kwargs["process_group"] = process_group
+
+        start_new_session = True if process_group is None else False
+
+        if not extra_kwargs:
+            return await anyio.open_process(
+                [command, *args],
+                env=env,
+                stderr=errlog,
+                cwd=cwd,
+                start_new_session=start_new_session,
+            )
+
+        try:
+            return await anyio.open_process(
+                [command, *args],
+                env=env,
+                stderr=errlog,
+                cwd=cwd,
+                start_new_session=start_new_session,
+                **extra_kwargs,
+            )
+        except TypeError:
+            backend: Any = getattr(anyio.lowlevel, "get_async_backend")()
+            return await backend.open_process(
+                [command, *args],
+                stdin=subprocess.PIPE,
+                stdout=subprocess.PIPE,
+                stderr=errlog,
+                cwd=cwd,
+                env=env,
+                start_new_session=start_new_session,
+                **extra_kwargs,
+            )
 
 
 async def _aclose_all(*streams: AsyncResource) -> None:

tests/client/test_stdio.py

@@ -17,7 +17,8 @@
 from collections.abc import Callable
 from contextlib import AsyncExitStack, suppress
 from pathlib import Path
-from typing import TextIO, cast
+from typing import Any, TextIO, cast
+from unittest.mock import MagicMock
 
 import anyio
 import anyio.abc
@@ -1203,8 +1204,12 @@ async def recording_spawn(
         env: dict[str, str] | None = None,
         errlog: TextIO = sys.stderr,
         cwd: Path | str | None = None,
+        *spawn_args: Any,
+        **spawn_kwargs: Any,
     ) -> anyio.abc.Process | FallbackProcess:
-        process = await _create_platform_compatible_process(command, args, env, errlog, cwd)
+        process = await _create_platform_compatible_process(
+            command, args, env, errlog, cwd, *spawn_args, **spawn_kwargs
+        )
         spawned.append(process)
         return process
 
@@ -1406,3 +1411,100 @@ async def test_a_graceful_exit_with_a_surviving_child_leaks_no_pipe_fds(  # prag
         # Subset, not equality: other machinery may close fds, but never open new
         # ones; a leaked pipe fd would show up as an extra entry.
         assert set(os.listdir("/proc/self/fd")) <= baseline
+
+
+def test_stdio_server_parameters_preexec_and_process_group() -> None:
+    """StdioServerParameters accepts preexec_fn and process_group with appropriate defaults."""
+    # Defaults
+    params = StdioServerParameters(command="echo")
+    assert params.preexec_fn is None
+    assert params.process_group is None
+
+    # Custom callable and process group
+    def hook() -> None:
+        pass
+
+    params_custom = StdioServerParameters(
+        command="echo",
+        preexec_fn=hook,
+        process_group=123,
+    )
+    assert params_custom.preexec_fn is hook
+    assert params_custom.process_group == 123
+
+
+@pytest.mark.anyio
+async def test_create_platform_compatible_process_forwards_preexec_and_process_group(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """_create_platform_compatible_process forwards preexec_fn and process_group to anyio.open_process."""
+    captured_kwargs: dict[str, Any] = {}
+
+    async def mock_open_process(*args: Any, **kwargs: Any) -> anyio.abc.Process:
+        captured_kwargs.update(kwargs)
+        return cast(anyio.abc.Process, MagicMock())
+
+    monkeypatch.setattr(anyio, "open_process", mock_open_process)
+
+    def dummy_preexec() -> None:
+        pass
+
+    await _create_platform_compatible_process(
+        "test-command",
+        ["--flag"],
+        preexec_fn=dummy_preexec,
+        process_group=12345,
+    )
+
+    assert captured_kwargs.get("preexec_fn") is dummy_preexec
+    assert captured_kwargs.get("process_group") == 12345
+    assert captured_kwargs.get("start_new_session") is False
+
+
+@pytest.mark.anyio
+@pytest.mark.skipif(sys.platform == "win32", reason="preexec_fn is POSIX only")
+async def test_preexec_fn_executes_in_child_process(tmp_path: Path) -> None:
+    """preexec_fn is called and executes in the child process before exec."""
+    log_file = tmp_path / "preexec.log"
+
+    def hook() -> None:
+        with open(log_file, "w") as f:
+            f.write(f"{os.getpid()}")
+
+    server_params = StdioServerParameters(
+        command=sys.executable,
+        args=["-c", "import sys; sys.stdin.read()"],
+        preexec_fn=hook,
+    )
+
+    with anyio.fail_after(5.0):
+        async with stdio_client(server_params):
+            while not log_file.exists():
+                await anyio.sleep(0.01)
+            child_pid = int(log_file.read_text().strip())
+            assert child_pid != os.getpid()
+
+
+@pytest.mark.anyio
+@pytest.mark.skipif(sys.platform == "win32", reason="process_group is POSIX only")
+async def test_process_group_sets_child_process_group(tmp_path: Path) -> None:
+    """process_group is passed and configured in the child process."""
+    pgid_file = tmp_path / "pgid.log"
+    script = (
+        f"import os, pathlib, sys\n"
+        f"pathlib.Path({str(pgid_file)!r}).write_text(f'{{os.getpid()}}:{{os.getpgrp()}}')\n"
+        f"sys.stdin.read()\n"
+    )
+    server_params = StdioServerParameters(
+        command=sys.executable,
+        args=["-c", script],
+        process_group=0,
+    )
+
+    with anyio.fail_after(5.0):
+        async with stdio_client(server_params):
+            while not pgid_file.exists():
+                await anyio.sleep(0.01)
+            pid_str, pgid_str = pgid_file.read_text().strip().split(":")
+            assert pid_str == pgid_str
+            assert int(pid_str) != os.getpid()