Skip to content

utilities

Windows-specific functionality for stdio client operations.

get_windows_executable_command 

get_windows_executable_command(command: str) -> str

Get the correct executable command normalized for Windows.

On Windows, commands might exist with specific extensions (.exe, .cmd, etc.) that need to be located for proper execution.

Parameters:

Name Type Description Default
command str

Base command (e.g., 'uvx', 'npx')

 required

Returns:

Name Type Description
str str

Windows-appropriate command path
Source code in src/mcp/os/win32/utilities.py 
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
def get_windows_executable_command(command: str) -> str:
    """Get the correct executable command normalized for Windows.

    On Windows, commands might exist with specific extensions (.exe, .cmd, etc.)
    that need to be located for proper execution.

    Args:
        command: Base command (e.g., 'uvx', 'npx')

    Returns:
        str: Windows-appropriate command path
    """
    try:
        # First check if command exists in PATH as-is
        if command_path := shutil.which(command):
            return command_path

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

        # For regular commands or if we couldn't find special versions
        return command
    except OSError:
        # Handle file system errors during path resolution
        # (permissions, broken symlinks, etc.)
        return command

FallbackProcess

A fallback process wrapper for Windows to handle async I/O when using subprocess.Popen, which provides sync-only FileIO objects.

This wraps stdin and stdout into async-compatible streams (FileReadStream, FileWriteStream), so that MCP clients expecting async streams can work properly.

Source code in src/mcp/os/win32/utilities.py 
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
class FallbackProcess:
    """A fallback process wrapper for Windows to handle async I/O
    when using subprocess.Popen, which provides sync-only FileIO objects.

    This wraps stdin and stdout into async-compatible
    streams (FileReadStream, FileWriteStream),
    so that MCP clients expecting async streams can work properly.
    """

    def __init__(self, popen_obj: subprocess.Popen[bytes]):
        self.popen: subprocess.Popen[bytes] = popen_obj
        self.stdin_raw = popen_obj.stdin  # type: ignore[assignment]
        self.stdout_raw = popen_obj.stdout  # type: ignore[assignment]
        self.stderr = popen_obj.stderr  # type: ignore[assignment]

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

    async def __aenter__(self):
        """Support async context manager entry."""
        return self

    async def __aexit__(
        self,
        exc_type: BaseException | None,
        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)

        # Close the file handles to prevent ResourceWarning
        if self.stdin:
            await self.stdin.aclose()
        if self.stdout:
            await self.stdout.aclose()
        if self.stdin_raw:
            self.stdin_raw.close()
        if self.stdout_raw:
            self.stdout_raw.close()
        if self.stderr:
            self.stderr.close()

    async def wait(self):
        """Async wait for process completion."""
        return await to_thread.run_sync(self.popen.wait)

    def terminate(self):
        """Terminate the subprocess immediately."""
        return self.popen.terminate()

    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

    @property
    def returncode(self) -> int | None:
        """Return the exit code, or ``None`` if the process has not yet terminated."""
        return self.popen.returncode

__aenter__ async 

__aenter__()

Support async context manager entry.

Source code in src/mcp/os/win32/utilities.py 
83
84
85
async def __aenter__(self):
    """Support async context manager entry."""
    return self

__aexit__ async 

__aexit__(
    exc_type: BaseException | None,
    exc_val: BaseException | None,
    exc_tb: object | None,
) -> None

Terminate and wait on process exit inside a thread.

Source code in src/mcp/os/win32/utilities.py 
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
async def __aexit__(
    self,
    exc_type: BaseException | None,
    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)

    # Close the file handles to prevent ResourceWarning
    if self.stdin:
        await self.stdin.aclose()
    if self.stdout:
        await self.stdout.aclose()
    if self.stdin_raw:
        self.stdin_raw.close()
    if self.stdout_raw:
        self.stdout_raw.close()
    if self.stderr:
        self.stderr.close()

wait async 

wait()

Async wait for process completion.

Source code in src/mcp/os/win32/utilities.py 
109
110
111
async def wait(self):
    """Async wait for process completion."""
    return await to_thread.run_sync(self.popen.wait)

terminate 

terminate()

Terminate the subprocess immediately.

Source code in src/mcp/os/win32/utilities.py 
113
114
115
def terminate(self):
    """Terminate the subprocess immediately."""
    return self.popen.terminate()

kill 

kill() -> None

Kill the subprocess immediately (alias for terminate).

Source code in src/mcp/os/win32/utilities.py 
117
118
119
def kill(self) -> None:
    """Kill the subprocess immediately (alias for terminate)."""
    self.terminate()

pid property 

pid: int

Return the process ID.

returncode property 

returncode: int | None

Return the exit code, or None if the process has not yet terminated.

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 in a Windows-compatible way with Job Object support.

Attempts 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.

Parameters:

Name Type Description Default
command str

The executable to run

 required
args list[str]

List of command line arguments

 required
env dict[str, str] | None

Environment variables

 None
errlog TextIO | None

Where to send stderr output (defaults to sys.stderr)

 stderr
cwd Path | str | None

Working directory for the subprocess

 None

Returns:

Type Description
Process | FallbackProcess

Process | FallbackProcess: Async-compatible subprocess with stdin and stdout streams
Source code in src/mcp/os/win32/utilities.py 
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
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 in a Windows-compatible way with Job Object support.

    Attempts 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
        env (dict[str, str] | None): Environment variables
        errlog (TextIO | None): Where to send stderr output (defaults to sys.stderr)
        cwd (Path | str | None): Working directory for the subprocess

    Returns:
        Process | FallbackProcess: Async-compatible subprocess with stdin and stdout streams
    """
    job = _create_job_object()
    process = None

    try:
        # First try using anyio with Windows-specific flags to hide console window
        process = await anyio.open_process(
            [command, *args],
            env=env,
            # Ensure we don't create console windows for each process
            creationflags=subprocess.CREATE_NO_WINDOW  # type: ignore
            if hasattr(subprocess, "CREATE_NO_WINDOW")
            else 0,
            stderr=errlog,
            cwd=cwd,
        )
    except NotImplementedError:
        # If Windows doesn't support async subprocess creation, use fallback
        process = await _create_windows_fallback_process(command, args, env, errlog, cwd)
    except Exception:
        # Try again without creation flags
        process = await anyio.open_process(
            [command, *args],
            env=env,
            stderr=errlog,
            cwd=cwd,
        )

    _maybe_assign_process_to_job(process, job)
    return process

terminate_windows_process_tree async 

terminate_windows_process_tree(
    process: Process | FallbackProcess,
    timeout_seconds: float = 2.0,
) -> 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.

Parameters:

Name Type Description Default
process Process | FallbackProcess

The process to terminate

 required
timeout_seconds float

Timeout in seconds before force killing (default: 2.0)

 2.0
Source code in src/mcp/os/win32/utilities.py 
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
async def terminate_windows_process_tree(process: Process | FallbackProcess, timeout_seconds: float = 2.0) -> 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
        timeout_seconds: Timeout in seconds before force killing (default: 2.0)
    """
    if sys.platform != "win32":
        return

    job = getattr(process, "_job_object", None)
    if job and win32job:
        try:
            win32job.TerminateJobObject(job, 1)
        except Exception:
            # Job might already be terminated
            pass
        finally:
            if win32api:
                try:
                    win32api.CloseHandle(job)
                except Exception:
                    pass

    # Always try to terminate the process itself as well
    try:
        process.terminate()
    except Exception:
        pass

terminate_windows_process async 

terminate_windows_process(
    process: Process | FallbackProcess,
)

Terminate a Windows process.

Note: On Windows, terminating a process with process.terminate() doesn't always guarantee immediate process termination. If the process does not exit within 2 seconds, process.kill() is called to send a SIGKILL-equivalent signal.

Parameters:

Name Type Description Default
process Process | FallbackProcess

The process to terminate

 required
Source code in src/mcp/os/win32/utilities.py 
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
@deprecated(
    "terminate_windows_process is deprecated and will be removed in a future version. "
    "Process termination is now handled internally by the stdio_client context manager."
)
async def terminate_windows_process(process: Process | FallbackProcess):
    """Terminate a Windows process.

    Note: On Windows, terminating a process with process.terminate() doesn't
    always guarantee immediate process termination.
    If the process does not exit within 2 seconds, process.kill() is called
    to send a SIGKILL-equivalent signal.

    Args:
        process: The process to terminate
    """
    try:
        process.terminate()
        with anyio.fail_after(2.0):
            await process.wait()
    except TimeoutError:
        # Force kill if it doesn't terminate
        process.kill()