Add deterministic explicit-start drag support

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.

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:

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})."

src/windows_mcp/uia/core.py

@@ -478,13 +478,36 @@ 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,
     x2: int,
     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)
 
 

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)]