utilities

Windows-specific functionality for stdio client operations.

get_windows_executable_command

get_windows_executable_command(command: str) -> str

Resolves the command to a Windows executable path.

Tries the bare name first, then the common script extensions (.cmd, .bat, .exe, .ps1).

Source code in src/mcp/os/win32/utilities.py

def get_windows_executable_command(command: str) -> str:
    """Resolves the command to a Windows executable path.

    Tries the bare name first, then the common script extensions (.cmd, .bat,
    .exe, .ps1).
    """
    try:
        if command_path := shutil.which(command):
            return command_path

        for ext in [".cmd", ".bat", ".exe", ".ps1"]:
            ext_version = f"{command}{ext}"
            if ext_path := shutil.which(ext_version):
                return ext_path

        return command
    except OSError:
        return command  # path probing failed (permissions, broken symlinks)

FallbackProcess

Async wrapper around subprocess.Popen for SelectorEventLoop.

Windows event loops without async subprocess support get this Popen-backed fallback, with anyio file streams wrapping the pipes.

Source code in src/mcp/os/win32/utilities.py

class FallbackProcess:
    """Async wrapper around subprocess.Popen for SelectorEventLoop.

    Windows event loops without async subprocess support get this Popen-backed
    fallback, with anyio file streams wrapping the pipes.
    """

    def __init__(self, popen_obj: subprocess.Popen[bytes]) -> None:
        self.popen: subprocess.Popen[bytes] = popen_obj
        stdin = popen_obj.stdin
        stdout = popen_obj.stdout

        self.stdin = FileWriteStream(cast(BinaryIO, stdin)) if stdin else None
        self.stdout = FileReadStream(cast(BinaryIO, stdout)) if stdout else None

    async def wait(self) -> int:
        """Waits for exit by polling the Popen.

        A thread blocked in Popen.wait() cannot be cancelled by anyio, which
        would defeat every timeout placed around this call.
        """
        while (returncode := self.popen.poll()) is None:
            await anyio.sleep(_EXIT_POLL_INTERVAL)
        return returncode

    def terminate(self) -> None:
        """Terminates the subprocess."""
        self.popen.terminate()

    def kill(self) -> None:
        """Kills the subprocess (on Windows the same hard kill as terminate)."""
        self.popen.kill()

    @property
    def pid(self) -> int:
        """Returns the process ID."""
        return self.popen.pid

    @property
    def returncode(self) -> int | None:
        """The exit code, or None while the process is still running.

        Polls the Popen so death is observable without anyone calling wait().
        """
        return self.popen.poll()

wait async

wait() -> int

Waits for exit by polling the Popen.

A thread blocked in Popen.wait() cannot be cancelled by anyio, which would defeat every timeout placed around this call.

Source code in src/mcp/os/win32/utilities.py

async def wait(self) -> int:
    """Waits for exit by polling the Popen.

    A thread blocked in Popen.wait() cannot be cancelled by anyio, which
    would defeat every timeout placed around this call.
    """
    while (returncode := self.popen.poll()) is None:
        await anyio.sleep(_EXIT_POLL_INTERVAL)
    return returncode

terminate

terminate() -> None

Terminates the subprocess.

Source code in src/mcp/os/win32/utilities.py

def terminate(self) -> None:
    """Terminates the subprocess."""
    self.popen.terminate()

kill

kill() -> None

Kills the subprocess (on Windows the same hard kill as terminate).

Source code in src/mcp/os/win32/utilities.py

def kill(self) -> None:
    """Kills the subprocess (on Windows the same hard kill as terminate)."""
    self.popen.kill()

pid property

pid: int

Returns the process ID.

returncode property

returncode: int | None

The exit code, or None while the process is still running.

Polls the Popen so death is observable without anyone calling wait().

create_windows_process async

create_windows_process(
    command: str,
    args: list[str],
    env: dict[str, str] | None = None,
    errlog: TextIO | None = stderr,
    cwd: Path | str | None = None,
) -> Process | FallbackProcess

Creates a subprocess with Job Object support for tree termination.

Spawns via anyio's open_process; event loops without async subprocess support (notably the SelectorEventLoop) raise NotImplementedError, in which case the spawn falls back to a Popen-backed FallbackProcess. Either way the process is then assigned to a Job Object so its children can be terminated with it; children spawned before the assignment completes are not captured (see the inline note below).

Returns:

Type	Description
Process | FallbackProcess	Process | FallbackProcess: The spawned process with async stdin/stdout streams.

Source code in src/mcp/os/win32/utilities.py

async def create_windows_process(
    command: str,
    args: list[str],
    env: dict[str, str] | None = None,
    errlog: TextIO | None = sys.stderr,
    cwd: Path | str | None = None,
) -> Process | FallbackProcess:
    """Creates a subprocess with Job Object support for tree termination.

    Spawns via anyio's open_process; event loops without async subprocess
    support (notably the SelectorEventLoop) raise NotImplementedError, in which
    case the spawn falls back to a Popen-backed FallbackProcess. Either way the
    process is then assigned to a Job Object so its children can be terminated
    with it; children spawned before the assignment completes are not captured
    (see the inline note below).

    Returns:
        Process | FallbackProcess: The spawned process with async stdin/stdout streams.
    """
    try:
        process = await anyio.open_process(
            [command, *args],
            env=env,
            # Ensure we don't create console windows for each process
            creationflags=getattr(subprocess, "CREATE_NO_WINDOW", 0),
            stderr=errlog,
            cwd=cwd,
        )
    except NotImplementedError:
        # Windows event loops without async subprocess support (SelectorEventLoop)
        process = await _create_windows_fallback_process(command, args, env, errlog, cwd)

    # Children spawned before the assignment completes land outside the job
    # (membership is inherited at CreateProcess, never acquired retroactively);
    # if that ever bites, the fix is a CREATE_SUSPENDED spawn -> assign -> resume.
    job = _create_job_object()
    _maybe_assign_process_to_job(process, job)
    return process

close_process_job

close_process_job(
    process: Process | FallbackProcess,
) -> None

Closes the process's Job Object handle, if it still has one.

KILL_ON_JOB_CLOSE makes the close also kill any members still alive, deterministically rather than at GC time; a deliberate divergence from POSIX, where a graceful server's children are left alive.

Source code in src/mcp/os/win32/utilities.py

def close_process_job(process: Process | FallbackProcess) -> None:
    """Closes the process's Job Object handle, if it still has one.

    KILL_ON_JOB_CLOSE makes the close also kill any members still alive,
    deterministically rather than at GC time; a deliberate divergence from
    POSIX, where a graceful server's children are left alive.
    """
    if sys.platform != "win32":
        return

    job = _process_jobs.pop(process, None)
    if job is not None:
        _close_job_handle(job)

terminate_windows_process_tree async

terminate_windows_process_tree(
    process: Process | FallbackProcess,
) -> None

Terminates the process's job, or just the process if it has no job.

Job termination is an immediate hard kill of every member. Windows has no tree-wide SIGTERM; the stdin-close grace period is the server's chance to exit cleanly.

Source code in src/mcp/os/win32/utilities.py

async def terminate_windows_process_tree(process: Process | FallbackProcess) -> None:
    """Terminates the process's job, or just the process if it has no job.

    Job termination is an immediate hard kill of every member. Windows has no
    tree-wide SIGTERM; the stdin-close grace period is the server's chance to
    exit cleanly.
    """
    if sys.platform != "win32":
        return

    job = _process_jobs.pop(process, None)
    if job is not None and win32job:
        try:
            with suppress(pywintypes.error):  # the job might already be terminated
                win32job.TerminateJobObject(job, 1)
        finally:
            _close_job_handle(job)

    # The process may have no job (creation or assignment failed); kill it directly too.
    try:
        process.terminate()
    except OSError:
        pass