From c2fc3e2fa03d1f906c8d4dc22ceac075262f03b3 Mon Sep 17 00:00:00 2001 From: Zeng Fanfan <8079595+zengfanfan@users.noreply.github.com> Date: Fri, 10 Jul 2026 21:09:01 +0800 Subject: [PATCH] Add deterministic Move drag options --- README.md | 6 +- src/windows_mcp/desktop/service.py | 87 +++++++++++++++- src/windows_mcp/tools/input.py | 44 ++++++++- src/windows_mcp/uia/core.py | 36 ++++++- tests/test_deterministic_drag_core.py | 72 ++++++++++++++ tests/test_deterministic_drag_desktop.py | 121 +++++++++++++++++++++++ tests/test_deterministic_drag_input.py | 101 +++++++++++++++++++ 7 files changed, 457 insertions(+), 10 deletions(-) create mode 100644 tests/test_deterministic_drag_core.py create mode 100644 tests/test_deterministic_drag_desktop.py create mode 100644 tests/test_deterministic_drag_input.py diff --git a/README.md b/README.md index b0b7f089..393f3607 100755 --- a/README.md +++ b/README.md @@ -691,7 +691,11 @@ MCP Client can access the following tools to interact with Windows: - `Click`: Click on the screen at the given coordinates. - `Type`: Type text on an element (optionally clears existing text). - `Scroll`: Scroll vertically or horizontally on the window or specific regions. -- `Move`: Move mouse pointer or drag (set drag=True) to coordinates. +- `Move`: Move mouse pointer or drag (set drag=True) to coordinates. For deterministic + drag, set `from_loc=[x, y]` with `drag=True` to press at an explicit start point and + release at `loc` in one tool call. Optional `duration` adds bounded intermediate + movement, and `expected_window_title` / `expected_process` fail before input if the + foreground target does not match. - `Shortcut`: Press keyboard shortcuts (`Ctrl+c`, `Alt+Tab`, etc). - `Wait`: Pause for a defined duration. - `WaitFor`: Wait until text, an active window, an element, or a focused element appears by polling UI state inside one tool call. diff --git a/src/windows_mcp/desktop/service.py b/src/windows_mcp/desktop/service.py index 9ad828b1..093a3a78 100755 --- a/src/windows_mcp/desktop/service.py +++ b/src/windows_mcp/desktop/service.py @@ -23,6 +23,7 @@ from fuzzywuzzy import process from time import sleep, time, perf_counter from psutil import Process +import math import win32process import win32gui import win32con @@ -770,14 +771,90 @@ def scroll( return 'Invalid type. Use "horizontal" or "vertical".' return None - def drag(self, loc: tuple[int, int] | list[int]): + def _normalize_drag_duration(self, duration: float | int | str | None) -> float | None: + if duration is None: + return None + try: + effective_duration = float(duration) + except (TypeError, ValueError) as exc: + raise ValueError("duration must be a finite number of seconds") from exc + if not math.isfinite(effective_duration): + raise ValueError("duration must be a finite number of seconds") + if effective_duration < 0 or effective_duration > 10: + raise ValueError("duration must be between 0 and 10 seconds") + return effective_duration + + def _assert_foreground_target( + self, + expected_window_title: str | None = None, + expected_process: str | None = None, + ) -> dict[str, object] | None: + if expected_window_title is None and expected_process is None: + return None + + active_window = self.get_foreground_window() + if active_window is None: + raise ValueError("No foreground window is available for target validation") + + title = active_window.Name or "" + process_name = "" + if expected_process is not None: + try: + process_name = Process(active_window.ProcessId).name() + except Exception as exc: + raise ValueError("Failed to resolve foreground process name") from exc + + if ( + expected_window_title is not None + and expected_window_title.casefold() not in title.casefold() + ): + raise ValueError( + "Foreground window title did not match expected_window_title: " + f"expected substring {expected_window_title!r}, actual {title!r}" + ) + + if expected_process is not None: + expected_basename = os.path.basename(expected_process).casefold() + if process_name.casefold() != expected_basename: + raise ValueError( + "Foreground process did not match expected_process: " + f"expected {expected_basename!r}, actual {process_name!r}" + ) + + return { + "title": title, + "process": process_name or None, + "process_id": active_window.ProcessId, + } + + def drag( + self, + loc: tuple[int, int] | list[int], + from_loc: tuple[int, int] | list[int] | None = None, + duration: float | int | str | None = None, + expected_window_title: str | None = None, + expected_process: str | None = None, + ) -> dict[str, object]: if isinstance(loc, list): x, y = loc[0], loc[1] else: x, y = loc + foreground = self._assert_foreground_target(expected_window_title, expected_process) + effective_duration = self._normalize_drag_duration(duration) sleep(0.5) - cx, cy = uia.GetCursorPos() - uia.DragDrop(cx, cy, x, y, moveSpeed=1) + if from_loc is None: + cx, cy = uia.GetCursorPos() + elif isinstance(from_loc, list): + cx, cy = from_loc[0], from_loc[1] + else: + cx, cy = from_loc + uia.DragDrop(cx, cy, x, y, moveSpeed=1, duration=effective_duration) + return { + "start": [cx, cy], + "end": [x, y], + "duration": effective_duration, + "foreground": foreground, + } def move(self, loc: tuple[int, int]): x, y = loc @@ -931,7 +1008,9 @@ def get_active_window(self, windows: list[Window] | None = None) -> Window | Non sleep(self._UIA_RETRY_SLEEP_MS / 1000.0) continue if last_error is not None: - logger.error(f"Error in get_active_window after {self._UIA_RETRIES} retries: {last_error}") + logger.error( + f"Error in get_active_window after {self._UIA_RETRIES} retries: {last_error}" + ) return None def get_foreground_window(self) -> uia.Control | None: diff --git a/src/windows_mcp/tools/input.py b/src/windows_mcp/tools/input.py index a8e748a1..5246f2d6 100644 --- a/src/windows_mcp/tools/input.py +++ b/src/windows_mcp/tools/input.py @@ -301,7 +301,10 @@ def scroll_tool( description=( "Moves mouse cursor to coordinates [x, y] or passing a UI element's label/id. " "Set drag=True to perform a drag-and-drop operation from the current mouse position " - "to the target coordinates. Default (drag=False) is a simple cursor move (hover). " + "to the target coordinates, or provide from_loc=[x, y] to make the drag explicit-start " + "and atomic in one tool call. Optional duration controls bounded intermediate movement. " + "Optional expected_window_title and expected_process guards fail before input if the " + "foreground target does not match. Default (drag=False) is a simple cursor move (hover). " "Provide either loc or label." ), annotations=ToolAnnotations( @@ -317,21 +320,54 @@ def move_tool( loc: list[int] | str | None = None, label: int | None = None, drag: bool | str = False, + from_loc: list[int] | str | None = None, + duration: float | int | str | None = None, + expected_window_title: str | None = None, + expected_process: str | None = None, ctx: Context = None, ) -> str: desktop = get_desktop() loc = _as_loc(loc) - drag = drag is True or (isinstance(drag, str) and drag.lower() == "true") + from_loc = _as_loc(from_loc) + drag = _as_bool(drag) if loc is None and label is None: raise ValueError("Either loc or label must be provided.") if label is not None: loc = _resolve_label(desktop, label) if len(loc) != 2: raise ValueError("loc must be a list of exactly 2 integers [x, y]") + if from_loc is not None and len(from_loc) != 2: + raise ValueError("from_loc must be a list of exactly 2 integers [x, y]") + has_drag_only_options = any( + value is not None + for value in ( + from_loc, + duration, + expected_window_title, + expected_process, + ) + ) + if has_drag_only_options and not drag: + raise ValueError( + "from_loc, duration, expected_window_title, and expected_process require drag=True" + ) x, y = loc[0], loc[1] if drag: - desktop.drag(loc) - return f"Dragged to ({x},{y})." + result = desktop.drag( + loc, + from_loc=from_loc, + duration=duration, + expected_window_title=expected_window_title, + expected_process=expected_process, + ) + start_x, start_y = result["start"] + effective_duration = result["duration"] + if effective_duration is None: + return f"Dragged from ({start_x},{start_y}) to ({x},{y})." + return ( + f"Dragged from ({start_x},{start_y}) to ({x},{y}) " + f"over {effective_duration:.3f} seconds." + ) else: desktop.move(loc) return f"Moved the mouse pointer to ({x},{y})." diff --git a/src/windows_mcp/uia/core.py b/src/windows_mcp/uia/core.py index 2b0433cc..6a983f4d 100755 --- a/src/windows_mcp/uia/core.py +++ b/src/windows_mcp/uia/core.py @@ -478,6 +478,28 @@ def MoveTo(x: int, y: int, moveSpeed: float = 1, waitTime: float = OPERATION_WAI time.sleep(waitTime) +def MoveToDuration(x: int, y: int, duration: float, waitTime: float = OPERATION_WAIT_TIME) -> None: + """ + Simulate mouse move to point x, y over a bounded duration from current cursor. + """ + curX, curY = GetCursorPos() + if duration <= 0: + SetCursorPos(x, y) + time.sleep(waitTime) + return + + stepCount = max(1, min(200, int(duration / 0.01))) + interval = duration / stepCount + for i in range(1, stepCount): + ratio = i / stepCount + cx = curX + round((x - curX) * ratio) + cy = curY + round((y - curY) * ratio) + SetCursorPos(cx, cy) + time.sleep(interval) + SetCursorPos(x, y) + time.sleep(waitTime) + + def DragDrop( x1: int, y1: int, @@ -485,6 +507,7 @@ def DragDrop( y2: int, moveSpeed: float = 1, waitTime: float = OPERATION_WAIT_TIME, + duration: float | None = None, ) -> None: """ Simulate mouse left button drag from point x1, y1 drop to point x2, y2. @@ -494,9 +517,20 @@ def DragDrop( y2: int. moveSpeed: float, 1 normal speed, < 1 move slower, > 1 move faster. waitTime: float. + duration: optional seconds for bounded intermediate movement. """ PressMouse(x1, y1, 0.05) - MoveTo(x2, y2, moveSpeed, 0.05) + try: + if duration is None: + MoveTo(x2, y2, moveSpeed, 0.05) + else: + MoveToDuration(x2, y2, duration, 0.05) + except BaseException as move_error: + try: + ReleaseMouse(waitTime) + except BaseException as release_error: + raise release_error from move_error + raise ReleaseMouse(waitTime) diff --git a/tests/test_deterministic_drag_core.py b/tests/test_deterministic_drag_core.py new file mode 100644 index 00000000..111023fb --- /dev/null +++ b/tests/test_deterministic_drag_core.py @@ -0,0 +1,72 @@ +import pytest + +from windows_mcp.uia import core + + +def test_dragdrop_releases_mouse_after_move_failure(monkeypatch: pytest.MonkeyPatch) -> None: + calls: list[str] = [] + + def press_mouse(x: int, y: int, wait_time: float) -> None: + calls.append(f"press:{x},{y},{wait_time}") + + def move_to(x: int, y: int, move_speed: float, wait_time: float) -> None: + calls.append(f"move:{x},{y},{move_speed},{wait_time}") + raise RuntimeError("move failed") + + def release_mouse(wait_time: float) -> None: + calls.append(f"release:{wait_time}") + + monkeypatch.setattr(core, "PressMouse", press_mouse) + monkeypatch.setattr(core, "MoveTo", move_to) + monkeypatch.setattr(core, "ReleaseMouse", release_mouse) + + with pytest.raises(RuntimeError, match="move failed"): + core.DragDrop(10, 20, 30, 40, moveSpeed=2, waitTime=0.3) + + assert calls == ["press:10,20,0.05", "move:30,40,2,0.05", "release:0.3"] + + +def test_dragdrop_uses_duration_path_and_releases(monkeypatch: pytest.MonkeyPatch) -> None: + calls: list[str] = [] + + monkeypatch.setattr(core, "PressMouse", lambda *args: calls.append(f"press:{args}")) + monkeypatch.setattr(core, "MoveTo", lambda *args: calls.append(f"move:{args}")) + monkeypatch.setattr(core, "MoveToDuration", lambda *args: calls.append(f"duration:{args}")) + monkeypatch.setattr(core, "ReleaseMouse", lambda *args: calls.append(f"release:{args}")) + + core.DragDrop(1, 2, 3, 4, moveSpeed=5, waitTime=0.6, duration=0.25) + + assert calls == [ + "press:(1, 2, 0.05)", + "duration:(3, 4, 0.25, 0.05)", + "release:(0.6,)", + ] + + +def test_move_to_duration_emits_intermediate_and_final_positions( + monkeypatch: pytest.MonkeyPatch, +) -> None: + positions: list[tuple[int, int]] = [] + + monkeypatch.setattr(core, "GetCursorPos", lambda: (0, 0)) + monkeypatch.setattr(core, "SetCursorPos", lambda x, y: positions.append((x, y))) + monkeypatch.setattr(core.time, "sleep", lambda seconds: None) + + core.MoveToDuration(30, 0, duration=0.04, waitTime=0) + + assert len(positions) > 1 + assert positions[-1] == (30, 0) + + +def test_move_to_duration_zero_duration_sets_final_position( + monkeypatch: pytest.MonkeyPatch, +) -> None: + positions: list[tuple[int, int]] = [] + + monkeypatch.setattr(core, "GetCursorPos", lambda: (10, 10)) + monkeypatch.setattr(core, "SetCursorPos", lambda x, y: positions.append((x, y))) + monkeypatch.setattr(core.time, "sleep", lambda seconds: None) + + core.MoveToDuration(10, 10, duration=0, waitTime=0) + + assert positions == [(10, 10)] diff --git a/tests/test_deterministic_drag_desktop.py b/tests/test_deterministic_drag_desktop.py new file mode 100644 index 00000000..db1249db --- /dev/null +++ b/tests/test_deterministic_drag_desktop.py @@ -0,0 +1,121 @@ +from types import SimpleNamespace + +import pytest + +from windows_mcp.desktop import service +from windows_mcp.desktop.service import Desktop + + +def _desktop() -> Desktop: + desktop = Desktop.__new__(Desktop) + desktop.desktop_state = None + return desktop + + +def test_desktop_drag_uses_explicit_start_duration_and_guards( + monkeypatch: pytest.MonkeyPatch, +) -> None: + calls: list[tuple[int, int, int, int, int, float | None]] = [] + desktop = _desktop() + + monkeypatch.setattr(service, "sleep", lambda seconds: None) + monkeypatch.setattr( + desktop, + "get_foreground_window", + lambda: SimpleNamespace(Name="Untitled - Notepad", ProcessId=123), + ) + monkeypatch.setattr(service, "Process", lambda pid: SimpleNamespace(name=lambda: "notepad.exe")) + monkeypatch.setattr( + service.uia, + "DragDrop", + lambda x1, y1, x2, y2, moveSpeed=1, duration=None: calls.append( + (x1, y1, x2, y2, moveSpeed, duration) + ), + ) + + result = desktop.drag( + [100, 200], + from_loc=[10, 20], + duration="0.25", + expected_window_title="notepad", + expected_process="NOTEPAD.EXE", + ) + + assert calls == [(10, 20, 100, 200, 1, 0.25)] + assert result["start"] == [10, 20] + assert result["end"] == [100, 200] + assert result["duration"] == 0.25 + assert result["foreground"] == { + "title": "Untitled - Notepad", + "process": "notepad.exe", + "process_id": 123, + } + + +def test_desktop_drag_legacy_start_uses_current_cursor(monkeypatch: pytest.MonkeyPatch) -> None: + calls: list[tuple[int, int, int, int]] = [] + desktop = _desktop() + + monkeypatch.setattr(service, "sleep", lambda seconds: None) + monkeypatch.setattr(service.uia, "GetCursorPos", lambda: (7, 8)) + monkeypatch.setattr( + service.uia, + "DragDrop", + lambda x1, y1, x2, y2, **kwargs: calls.append((x1, y1, x2, y2)), + ) + + result = desktop.drag((30, 40)) + + assert calls == [(7, 8, 30, 40)] + assert result["start"] == [7, 8] + assert result["duration"] is None + + +def test_desktop_drag_fails_before_press_on_title_mismatch( + monkeypatch: pytest.MonkeyPatch, +) -> None: + desktop = _desktop() + + monkeypatch.setattr(service, "sleep", lambda seconds: None) + monkeypatch.setattr( + desktop, + "get_foreground_window", + lambda: SimpleNamespace(Name="Calculator", ProcessId=123), + ) + monkeypatch.setattr( + service.uia, + "DragDrop", + lambda *args, **kwargs: (_ for _ in ()).throw(AssertionError("should not drag")), + ) + + with pytest.raises(ValueError, match="expected_window_title"): + desktop.drag([100, 200], from_loc=[10, 20], expected_window_title="Notepad") + + +def test_desktop_drag_fails_before_press_on_process_mismatch( + monkeypatch: pytest.MonkeyPatch, +) -> None: + desktop = _desktop() + + monkeypatch.setattr(service, "sleep", lambda seconds: None) + monkeypatch.setattr( + desktop, + "get_foreground_window", + lambda: SimpleNamespace(Name="Untitled - Notepad", ProcessId=123), + ) + monkeypatch.setattr(service, "Process", lambda pid: SimpleNamespace(name=lambda: "notepad.exe")) + monkeypatch.setattr( + service.uia, + "DragDrop", + lambda *args, **kwargs: (_ for _ in ()).throw(AssertionError("should not drag")), + ) + + with pytest.raises(ValueError, match="expected_process"): + desktop.drag([100, 200], from_loc=[10, 20], expected_process="note.exe") + + +def test_desktop_drag_rejects_non_finite_duration() -> None: + desktop = _desktop() + + with pytest.raises(ValueError, match="finite"): + desktop.drag([1, 2], from_loc=[3, 4], duration="nan") diff --git a/tests/test_deterministic_drag_input.py b/tests/test_deterministic_drag_input.py new file mode 100644 index 00000000..834440f9 --- /dev/null +++ b/tests/test_deterministic_drag_input.py @@ -0,0 +1,101 @@ +import asyncio +from collections.abc import Callable + +import pytest + +from windows_mcp.tools.input import register + + +class FakeMCP: + def __init__(self) -> None: + self.tools: dict[str, Callable] = {} + + def tool(self, *, name: str, **kwargs: object) -> Callable: + def decorator(func: Callable) -> Callable: + self.tools[name] = func + return func + + return decorator + + +class FakeDesktop: + def __init__(self) -> None: + self.desktop_state = object() + self.move_calls: list[list[int]] = [] + self.drag_calls: list[dict[str, object]] = [] + + def move(self, loc: list[int]) -> None: + self.move_calls.append(loc) + + def drag(self, loc: list[int], **kwargs: object) -> dict[str, object]: + self.drag_calls.append({"loc": loc, **kwargs}) + return { + "start": kwargs.get("from_loc") or [1, 2], + "end": loc, + "duration": 0.25 if kwargs.get("duration") is not None else None, + "foreground": None, + } + + +def _tools(desktop: FakeDesktop) -> dict[str, Callable]: + mcp = FakeMCP() + register(mcp, get_desktop=lambda: desktop, get_analytics=lambda: None) + return mcp.tools + + +def test_move_tool_preserves_legacy_move_behavior() -> None: + desktop = FakeDesktop() + result = asyncio.run(_tools(desktop)["Move"](loc=[10, 20])) + + assert result == "Moved the mouse pointer to (10,20)." + assert desktop.move_calls == [[10, 20]] + assert desktop.drag_calls == [] + + +def test_move_tool_accepts_explicit_drag_start_list() -> None: + desktop = FakeDesktop() + result = asyncio.run( + _tools(desktop)["Move"]( + loc=[100, 200], + drag=True, + from_loc=[10, 20], + duration=0.25, + expected_window_title="Notepad", + expected_process="notepad.exe", + ) + ) + + assert result == "Dragged from (10,20) to (100,200) over 0.250 seconds." + assert desktop.drag_calls == [ + { + "loc": [100, 200], + "from_loc": [10, 20], + "duration": 0.25, + "expected_window_title": "Notepad", + "expected_process": "notepad.exe", + } + ] + + +def test_move_tool_accepts_explicit_drag_start_json_string() -> None: + desktop = FakeDesktop() + result = asyncio.run( + _tools(desktop)["Move"]( + loc="[100, 200]", + drag="true", + from_loc="[10, 20]", + ) + ) + + assert result == "Dragged from (10,20) to (100,200)." + assert desktop.drag_calls[0]["from_loc"] == [10, 20] + + +def test_move_tool_rejects_invalid_from_loc() -> None: + with pytest.raises(ValueError, match="from_loc"): + asyncio.run(_tools(FakeDesktop())["Move"](loc=[100, 200], drag=True, from_loc=[10])) + + +def test_move_tool_rejects_drag_options_without_drag() -> None: + with pytest.raises(ValueError, match="require drag=True"): + asyncio.run(_tools(FakeDesktop())["Move"](loc=[100, 200], from_loc=[10, 20]))