Fix Windows subprocess NotImplementedError (STDIO clients)

README.md

@@ -315,42 +315,27 @@ async def long_task(files: list[str], ctx: Context) -> str:
 Authentication can be used by servers that want to expose tools accessing protected resources.
 
 `mcp.server.auth` implements an OAuth 2.0 server interface, which servers can use by
-providing an implementation of the `OAuthAuthorizationServerProvider` protocol.
+providing an implementation of the `OAuthServerProvider` protocol.
 
-```python
-from mcp import FastMCP
-from mcp.server.auth.provider import OAuthAuthorizationServerProvider
-from mcp.server.auth.settings import (
-    AuthSettings,
-    ClientRegistrationOptions,
-    RevocationOptions,
-)
-
-
-class MyOAuthServerProvider(OAuthAuthorizationServerProvider):
-    # See an example on how to implement at `examples/servers/simple-auth`
-    ...
-
-
-mcp = FastMCP(
-    "My App",
-    auth_server_provider=MyOAuthServerProvider(),
-    auth=AuthSettings(
-        issuer_url="https://myapp.com",
-        revocation_options=RevocationOptions(
-            enabled=True,
-        ),
-        client_registration_options=ClientRegistrationOptions(
-            enabled=True,
-            valid_scopes=["myscope", "myotherscope"],
-            default_scopes=["myscope"],
+```
+mcp = FastMCP("My App",
+        auth_server_provider=MyOAuthServerProvider(),
+        auth=AuthSettings(
+            issuer_url="https://myapp.com",
+            revocation_options=RevocationOptions(
+                enabled=True,
+            ),
+            client_registration_options=ClientRegistrationOptions(
+                enabled=True,
+                valid_scopes=["myscope", "myotherscope"],
+                default_scopes=["myscope"],
+            ),
+            required_scopes=["myscope"],
         ),
-        required_scopes=["myscope"],
-    ),
 )
 ```
 
-See [OAuthAuthorizationServerProvider](src/mcp/server/auth/provider.py) for more details.
+See [OAuthServerProvider](src/mcp/server/auth/provider.py) for more details.
 
 ## Running Your Server
 
@@ -477,12 +462,15 @@ For low level server with Streamable HTTP implementations, see:
 - Stateful server: [`examples/servers/simple-streamablehttp/`](examples/servers/simple-streamablehttp/)
 - Stateless server: [`examples/servers/simple-streamablehttp-stateless/`](examples/servers/simple-streamablehttp-stateless/)
 
+
+
 The streamable HTTP transport supports:
 - Stateful and stateless operation modes
 - Resumability with event stores
-- JSON or SSE response formats
+- JSON or SSE response formats  
 - Better scalability for multi-node deployments
 
+
 ### Mounting to an Existing ASGI Server
 
 > **Note**: SSE transport is being superseded by [Streamable HTTP transport](https://modelcontextprotocol.io/specification/2025-03-26/basic/transports#streamable-http).

src/mcp/client/stdio/__init__.py

@@ -108,28 +108,20 @@ async def stdio_client(server: StdioServerParameters, errlog: TextIO = sys.stder
     read_stream_writer, read_stream = anyio.create_memory_object_stream(0)
     write_stream, write_stream_reader = anyio.create_memory_object_stream(0)
 
-    try:
-        command = _get_executable_command(server.command)
-
-        # Open process with stderr piped for capture
-        process = await _create_platform_compatible_process(
-            command=command,
-            args=server.args,
-            env=(
-                {**get_default_environment(), **server.env}
-                if server.env is not None
-                else get_default_environment()
-            ),
-            errlog=errlog,
-            cwd=server.cwd,
-        )
-    except OSError:
-        # Clean up streams if process creation fails
-        await read_stream.aclose()
-        await write_stream.aclose()
-        await read_stream_writer.aclose()
-        await write_stream_reader.aclose()
-        raise
+    command = _get_executable_command(server.command)
+
+    # Open process with stderr piped for capture
+    process = await _create_platform_compatible_process(
+        command=command,
+        args=server.args,
+        env=(
+            {**get_default_environment(), **server.env}
+            if server.env is not None
+            else get_default_environment()
+        ),
+        errlog=errlog,
+        cwd=server.cwd,
+    )
 
     async def stdout_reader():
         assert process.stdout, "Opened process is missing stdout"
@@ -185,18 +177,12 @@ async def stdin_writer():
             yield read_stream, write_stream
         finally:
             # Clean up process to prevent any dangling orphaned processes
-            try:
-                if sys.platform == "win32":
-                    await terminate_windows_process(process)
-                else:
-                    process.terminate()
-            except ProcessLookupError:
-                # Process already exited, which is fine
-                pass
+            if sys.platform == "win32":
+                await terminate_windows_process(process)
+            else:
+                process.terminate()
             await read_stream.aclose()
             await write_stream.aclose()
-            await read_stream_writer.aclose()
-            await write_stream_reader.aclose()
 
 
 def _get_executable_command(command: str) -> str:

src/mcp/client/stdio/win32.py

@@ -6,10 +6,12 @@
 import subprocess
 import sys
 from pathlib import Path
-from typing import TextIO
+from typing import BinaryIO, TextIO, cast
 
 import anyio
+from anyio import to_thread
 from anyio.abc import Process
+from anyio.streams.file import FileReadStream, FileWriteStream
 
 
 def get_windows_executable_command(command: str) -> str:
@@ -44,48 +46,107 @@ def get_windows_executable_command(command: str) -> str:
         return command
 
 
+class DummyProcess:
+    """
+    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)
+
+    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()
+
+
+# ------------------------
+# Updated function
+# ------------------------
+
+
 async def create_windows_process(
     command: str,
     args: list[str],
     env: dict[str, str] | None = None,
-    errlog: TextIO = sys.stderr,
+    errlog: TextIO | None = sys.stderr,
     cwd: Path | str | None = None,
-):
+) -> DummyProcess:
     """
     Creates a subprocess in a Windows-compatible way.
 
-    Windows processes need special handling for console windows and
-    process creation flags.
+    On Windows, asyncio.create_subprocess_exec has incomplete support
+    (NotImplementedError when trying to open subprocesses).
+    Therefore, we fallback to subprocess.Popen and wrap it for async usage.
 
     Args:
-        command: The command to execute
-        args: Command line arguments
-        env: Environment variables
-        errlog: Where to send stderr output
-        cwd: Working directory for the process
+        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:
-        A process handle
+        DummyProcess: Async-compatible subprocess with stdin and stdout streams
     """
     try:
-        # Try with Windows-specific flags to hide console window
-        process = await anyio.open_process(
+        # Try launching with creationflags to avoid opening a new console window
+        popen_obj = subprocess.Popen(
             [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,
+            stdin=subprocess.PIPE,
+            stdout=subprocess.PIPE,
             stderr=errlog,
+            env=env,
             cwd=cwd,
+            bufsize=0,  # Unbuffered output
+            creationflags = getattr(subprocess, "CREATE_NO_WINDOW", 0),
         )
-        return process
+        return DummyProcess(popen_obj)
+
     except Exception:
-        # Don't raise, let's try to create the process without creation flags
-        process = await anyio.open_process(
-            [command, *args], env=env, stderr=errlog, cwd=cwd
+        # If creationflags failed, fallback without them
+        popen_obj = subprocess.Popen(
+            [command, *args],
+            stdin=subprocess.PIPE,
+            stdout=subprocess.PIPE,
+            stderr=errlog,
+            env=env,
+            cwd=cwd,
+            bufsize=0,
         )
-        return process
+        return DummyProcess(popen_obj)
 
 
 async def terminate_windows_process(process: Process):

tests/client/test_stdio.py

@@ -2,17 +2,11 @@
 
 import pytest
 
-from mcp.client.session import ClientSession
-from mcp.client.stdio import (
-    StdioServerParameters,
-    stdio_client,
-)
-from mcp.shared.exceptions import McpError
+from mcp.client.stdio import StdioServerParameters, stdio_client
 from mcp.shared.message import SessionMessage
-from mcp.types import CONNECTION_CLOSED, JSONRPCMessage, JSONRPCRequest, JSONRPCResponse
+from mcp.types import JSONRPCMessage, JSONRPCRequest, JSONRPCResponse
 
 tee: str = shutil.which("tee")  # type: ignore
-python: str = shutil.which("python")  # type: ignore
 
 
 @pytest.mark.anyio
@@ -56,43 +50,3 @@ async def test_stdio_client():
         assert read_messages[1] == JSONRPCMessage(
             root=JSONRPCResponse(jsonrpc="2.0", id=2, result={})
         )
-
-
-@pytest.mark.anyio
-async def test_stdio_client_bad_path():
-    """Check that the connection doesn't hang if process errors."""
-    server_params = StdioServerParameters(
-        command="python", args=["-c", "non-existent-file.py"]
-    )
-    async with stdio_client(server_params) as (read_stream, write_stream):
-        async with ClientSession(read_stream, write_stream) as session:
-            # The session should raise an error when the connection closes
-            with pytest.raises(McpError) as exc_info:
-                await session.initialize()
-
-            # Check that we got a connection closed error
-            assert exc_info.value.error.code == CONNECTION_CLOSED
-            assert "Connection closed" in exc_info.value.error.message
-
-
-@pytest.mark.anyio
-async def test_stdio_client_nonexistent_command():
-    """Test that stdio_client raises an error for non-existent commands."""
-    # Create a server with a non-existent command
-    server_params = StdioServerParameters(
-        command="/path/to/nonexistent/command",
-        args=["--help"],
-    )
-
-    # Should raise an error when trying to start the process
-    with pytest.raises(Exception) as exc_info:
-        async with stdio_client(server_params) as (_, _):
-            pass
-
-    # The error should indicate the command was not found
-    error_message = str(exc_info.value)
-    assert (
-        "nonexistent" in error_message
-        or "not found" in error_message.lower()
-        or "cannot find the file" in error_message.lower()  # Windows error message
-    )