Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions pyproject.toml
Original file line number Diff line number Diff line change
Expand Up @@ -77,6 +77,7 @@ dev = [
"strict-no-cover",
"logfire>=3.0.0",
"opentelemetry-sdk>=1.39.1",
"blockbuster>=1.5.27",
]
docs = [
# Zensical is the Material team's successor to MkDocs; it natively
Expand Down
9 changes: 5 additions & 4 deletions src/mcp/client/stdio.py
Original file line number Diff line number Diff line change
Expand Up @@ -18,6 +18,7 @@

import anyio
import anyio.lowlevel
import anyio.to_thread
import mcp_types as types
from anyio.abc import AsyncResource, Process
from anyio.streams.text import TextReceiveStream
Expand Down Expand Up @@ -120,7 +121,7 @@ async def stdio_client(
OSError: If the server process cannot be spawned.
ValueError: If the spawn parameters are invalid (embedded NUL bytes).
"""
command = _get_executable_command(server.command)
command = await _get_executable_command(server.command)

process = await _create_platform_compatible_process(
command=command,
Expand Down Expand Up @@ -317,10 +318,10 @@ def _close_subprocess_transport(process: ServerProcess) -> None:
close()


def _get_executable_command(command: str) -> str:
async def _get_executable_command(command: str) -> str:
"""Normalizes the command for the current platform."""
if sys.platform == "win32": # pragma: no cover
return get_windows_executable_command(command)
if sys.platform == "win32":
return await anyio.to_thread.run_sync(get_windows_executable_command, command, abandon_on_cancel=True)
else: # pragma: lax no cover
return command

Expand Down
47 changes: 47 additions & 0 deletions tests/client/test_stdio.py
Original file line number Diff line number Diff line change
Expand Up @@ -14,14 +14,18 @@
import os
import signal
import sys
import threading
from collections.abc import Callable
from contextlib import AsyncExitStack, suppress
from pathlib import Path
from types import SimpleNamespace
from typing import TextIO, cast

import anyio
import anyio.abc
import anyio.from_thread
import anyio.lowlevel
import anyio.to_thread
import pytest
import trio
import trio.testing
Expand Down Expand Up @@ -199,6 +203,9 @@ def install_fake_process(
"""
terminated: list[FakeProcess] = []

async def fake_get_executable_command(command: str) -> str:
return command

async def fake_spawn(
command: str,
args: list[str],
Expand All @@ -212,6 +219,7 @@ async def fake_terminate_tree(proc: FakeProcess) -> None:
terminated.append(proc)
proc.exit(-15)

monkeypatch.setattr(stdio, "_get_executable_command", fake_get_executable_command)
monkeypatch.setattr(stdio, "_create_platform_compatible_process", fake_spawn)
monkeypatch.setattr(stdio, "_terminate_process_tree", fake_terminate_tree)
if grace_period is not None:
Expand Down Expand Up @@ -568,6 +576,45 @@ async def test_a_command_that_cannot_be_execed_raises_enoent() -> None:
assert exc_info.value.errno == errno.ENOENT


@pytest.mark.anyio
async def test_cancellation_during_windows_command_resolution_returns_before_resolution_finishes(
monkeypatch: pytest.MonkeyPatch,
) -> None:
"""Cancelling `stdio_client` does not wait for blocked Windows command resolution."""
resolution_started = anyio.Event()
resolution_release = threading.Event()
resolution_finished = threading.Event()

def blocking_resolver(command: str) -> str:
anyio.from_thread.run_sync(resolution_started.set)
resolution_release.wait()
resolution_finished.set()
return command

monkeypatch.setattr(stdio, "sys", SimpleNamespace(platform="win32"))
monkeypatch.setattr(stdio, "get_windows_executable_command", blocking_resolver)

cancel_scope = anyio.CancelScope()
client_stopped = anyio.Event()

async def run_client() -> None:
with cancel_scope:
async with AsyncExitStack() as stack:
await stack.enter_async_context(stdio_client(FAKE_PARAMS))
client_stopped.set()

with anyio.fail_after(5):
async with anyio.create_task_group() as tg:
tg.start_soon(run_client)
await resolution_started.wait()
cancel_scope.cancel()
try:
await client_stopped.wait()
finally:
resolution_release.set()
await anyio.to_thread.run_sync(resolution_finished.wait)


@pytest.mark.anyio
async def test_cancellation_during_spawn_leaks_no_streams(monkeypatch: pytest.MonkeyPatch) -> None:
"""Cancellation while the spawn is still in flight must not leak the internal streams.
Expand Down
26 changes: 26 additions & 0 deletions tests/conftest.py
Original file line number Diff line number Diff line change
@@ -1,7 +1,9 @@
import os
from collections.abc import AsyncIterator, Iterator

import httpcore2 as _httpcore2
import pytest
from blockbuster import BlockBuster

# OpenTelemetry's `set_tracer_provider` is set-once per process, so the suite
# uses a single span-capture mechanism: logfire's `capfire` fixture (its
Expand All @@ -17,12 +19,36 @@

import mcp.shared._otel # noqa: E402

# Load httpx2's lazy default transport before BlockBuster starts.
del _httpcore2


@pytest.fixture(scope="session")
def anyio_backend() -> str:
return "asyncio"


@pytest.fixture(autouse=True)
def blockbuster() -> Iterator[None]:
bb = BlockBuster(["mcp", "mcp_types"])
# Coverage reads source files while collecting data.
bb.functions["os.stat"].can_block_in("coverage/python.py", "get_python_source")
bb.functions["io.BufferedReader.read"].can_block_in("coverage/python.py", "read_python_source")
# jsonschema discovers its bundled schemas during its first import.
bb.functions["os.listdir"].can_block_in("/jsonschema_specifications/_core.py", "_schemas")
bb.functions["os.scandir"].can_block_in("/jsonschema_specifications/_core.py", "_schemas")
bb.functions["io.TextIOWrapper.read"].can_block_in("/jsonschema_specifications/_core.py", "_schemas")
# These public synchronous conversions read the media file by design.
bb.functions["io.BufferedReader.read"].can_block_in(
"mcp/server/mcpserver/utilities/types.py", ("to_image_content", "to_audio_content")
)
bb.activate()
try:
yield
finally:
bb.deactivate()


@pytest.fixture(scope="module", autouse=True)
async def _module_runner_lease(anyio_backend: str) -> AsyncIterator[None]:
"""Share one event loop across each module's tests instead of one per test.
Expand Down
20 changes: 20 additions & 0 deletions uv.lock

Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.

Loading