From b160c5bcd69f4a41c5b547a672b5af6314b4dc60 Mon Sep 17 00:00:00 2001 From: laplace young Date: Thu, 21 May 2026 20:35:03 +0800 Subject: [PATCH] Reduce snapshot churn during UI readiness checks Agents can now wait for common UI readiness signals inside one tool call, using the existing desktop state capture path and a bounded polling loop. Constraint: Issue #87 discussion prioritizes condition waits with a small polling fallback. Rejected: Event-driven WatchDog wakeups in this PR | polling contract keeps the contribution narrow. Confidence: high Scope-risk: narrow Directive: Preserve WaitFor inputs and timeout semantics when adding WatchDog wakeups. Tested: uv run ruff check src/windows_mcp/tools/input.py tests/test_wait_for_tool.py README.md; uv run python -m pytest tests/test_wait_for_tool.py tests/test_multi_tools.py; uv run python -m pytest (304 passed, 6 existing tests/test_tree_views.py failures observed) Not-tested: Repository-wide green pytest run; current run reports the existing tests/test_tree_views.py failures. --- README.md | 1 + src/windows_mcp/tools/input.py | 232 ++++++++++++++++++++++++++++++++- tests/test_wait_for_tool.py | 201 ++++++++++++++++++++++++++++ 3 files changed, 429 insertions(+), 5 deletions(-) create mode 100644 tests/test_wait_for_tool.py diff --git a/README.md b/README.md index 06d02bb1..3db6408f 100755 --- a/README.md +++ b/README.md @@ -694,6 +694,7 @@ MCP Client can access the following tools to interact with Windows: - `Move`: Move mouse pointer or drag (set drag=True) to coordinates. - `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. - `Screenshot`: Fast screenshot-first desktop capture with cursor position, active/open windows, and an image. Skips UI tree extraction for speed and should be the default first call when you mainly need visual context. Supports `display=[0]` or `display=[0,1]` to capture specific screens. After capture, a brief orange-red glowing border is drawn over the captured area as a visual confirmation (set `WINDOWS_MCP_DISABLE_FLASH=1` to disable). - `Snapshot`: Full desktop state capture for workflows that need interactive element ids, scrollable regions, or `use_dom=True` browser extraction. Supports `use_vision=True` for including screenshots and `display=[0]` or `display=[0,1]` for limiting all returned Snapshot information to specific screens. - `App`: To launch an application from the start menu, resize or move the window and switch between apps. diff --git a/src/windows_mcp/tools/input.py b/src/windows_mcp/tools/input.py index 16ff11cc..47ca644d 100644 --- a/src/windows_mcp/tools/input.py +++ b/src/windows_mcp/tools/input.py @@ -1,14 +1,24 @@ -"""Input tools — Click, Type, Scroll, Move, Shortcut, Wait.""" +"""Input tools — Click, Type, Scroll, Move, Shortcut, Wait, WaitFor.""" import time -from typing import Literal +from collections.abc import Callable, Iterator +from typing import Any, Literal from mcp.types import ToolAnnotations from windows_mcp.infrastructure import with_analytics from fastmcp import Context -def _resolve_label(desktop, label): +WaitForCondition = Literal[ + "text_exists", + "active_window", + "element_exists", + "element_enabled", + "focused_element", +] + + +def _resolve_label(desktop: Any, label: int) -> list[int]: """Resolve a UI element label to screen coordinates.""" if desktop.desktop_state is None: raise ValueError("Desktop state is empty. Please call Snapshot first.") @@ -18,7 +28,150 @@ def _resolve_label(desktop, label): raise ValueError(f"Failed to find element with label {label}: {e}") -def register(mcp, *, get_desktop, get_analytics): +def _as_bool(value: bool | str) -> bool: + return value is True or (isinstance(value, str) and value.lower() == "true") + + +def _text_matches(value: object | None, expected: str | None) -> bool: + if expected is None: + return True + if value is None: + return False + return expected.casefold() in str(value).casefold() + + +def _metadata_text_matches(metadata: dict[str, object], expected: str | None) -> bool: + return any(_text_matches(value, expected) for value in metadata.values()) + + +def _iter_nodes(desktop_state: Any) -> Iterator[Any]: + tree_state = getattr(desktop_state, "tree_state", None) + if tree_state is None: + return + yield from getattr(tree_state, "interactive_nodes", []) + yield from getattr(tree_state, "scrollable_nodes", []) + + +def _iter_text_sources(desktop_state: Any) -> Iterator[object]: + active_window = getattr(desktop_state, "active_window", None) + if active_window is not None: + yield active_window.name + + for window in getattr(desktop_state, "windows", []): + yield window.name + + tree_state = getattr(desktop_state, "tree_state", None) + if tree_state is None: + return + + for node in _iter_nodes(desktop_state): + yield node.name + yield node.control_type + yield node.window_name + for value in getattr(node, "metadata", {}).values(): + yield value + + for node in getattr(tree_state, "dom_informative_nodes", []): + yield getattr(node, "text", "") + + +def _node_matches(node: Any, text: str | None, window_name: str | None) -> bool: + metadata: dict[str, object] = getattr(node, "metadata", {}) + return ( + _text_matches(getattr(node, "name", ""), text) + or _text_matches(getattr(node, "control_type", ""), text) + or _metadata_text_matches(metadata, text) + ) and _text_matches(getattr(node, "window_name", ""), window_name) + + +def _matches_wait_condition( + desktop_state: Any, + condition: WaitForCondition, + text: str | None, + window_name: str | None, +) -> tuple[bool, str]: + if condition == "text_exists": + for source in _iter_text_sources(desktop_state): + if _text_matches(source, text): + return True, f"text {text!r} appeared" + return False, f"text {text!r} was absent" + + if condition == "active_window": + expected = window_name or text + active_window = getattr(desktop_state, "active_window", None) + active_name = active_window.name if active_window else "" + if _text_matches(active_name, expected): + return True, f"active window matched {active_name!r}" + return False, f"active window was {active_name!r}" + + if condition in {"element_exists", "element_enabled"}: + for node in _iter_nodes(desktop_state): + if _node_matches(node, text, window_name): + return True, f"element matched {getattr(node, 'name', '')!r}" + return False, "matching element was absent" + + if condition == "focused_element": + for node in _iter_nodes(desktop_state): + metadata = getattr(node, "metadata", {}) + if metadata.get("has_focused") and _node_matches(node, text, window_name): + return True, f"focused element matched {getattr(node, 'name', '')!r}" + return False, "matching focused element was absent" + + raise ValueError(f"Unsupported WaitFor condition: {condition}") + + +def _validate_wait_for_args( + condition: str, + text: str | None, + window_name: str | None, + timeout: float, + interval: float, +) -> WaitForCondition: + normalized = condition.strip().lower().replace("-", "_") + aliases = { + "text": "text_exists", + "window": "active_window", + "element": "element_exists", + "enabled": "element_enabled", + "focused": "focused_element", + } + normalized = aliases.get(normalized, normalized) + valid_conditions = { + "text_exists", + "active_window", + "element_exists", + "element_enabled", + "focused_element", + } + if normalized not in valid_conditions: + raise ValueError( + "condition must be one of: text_exists, active_window, element_exists, " + "element_enabled, focused_element" + ) + + if timeout <= 0 or timeout > 120: + raise ValueError("timeout must be greater than 0 and at most 120 seconds") + if interval <= 0 or interval > 5: + raise ValueError("interval must be greater than 0 and at most 5 seconds") + + if normalized == "text_exists" and not text: + raise ValueError("text is required when condition is text_exists") + if normalized == "active_window" and not (text or window_name): + raise ValueError("text or window_name is required when condition is active_window") + if normalized in {"element_exists", "element_enabled"} and not (text or window_name): + raise ValueError( + "text or window_name is required when condition is element_exists or element_enabled" + ) + + return normalized + + +def register( + mcp: Any, + *, + get_desktop: Callable[[], Any], + get_analytics: Callable[[], Any], +) -> None: @mcp.tool( name="Click", description=( @@ -122,7 +275,8 @@ def scroll_tool( if response: return response return ( - f"Scrolled {type} {direction} by {wheel_times} wheel times" + f" at ({loc[0]},{loc[1]})." + f"Scrolled {type} {direction} by {wheel_times} wheel times" + + f" at ({loc[0]},{loc[1]})." if loc else "" ) @@ -197,3 +351,71 @@ def shortcut_tool(shortcut: str, ctx: Context = None): def wait_tool(duration: int, ctx: Context = None) -> str: time.sleep(duration) return f"Waited for {duration} seconds." + + @mcp.tool( + name="WaitFor", + description=( + "Waits until a UI condition is satisfied, polling the Windows accessibility tree " + "inside the tool to avoid repeated Snapshot calls. Conditions: text_exists, " + "active_window, element_exists, element_enabled, focused_element. Provide text " + "and/or window_name depending on the condition. Set use_dom=True for browser DOM text." + ), + annotations=ToolAnnotations( + title="WaitFor", + readOnlyHint=True, + destructiveHint=False, + idempotentHint=True, + openWorldHint=False, + ), + ) + @with_analytics(get_analytics(), "WaitFor-Tool") + def wait_for_tool( + condition: str, + text: str | None = None, + window_name: str | None = None, + timeout: float = 10.0, + interval: float = 0.25, + use_dom: bool | str = False, + ctx: Context = None, + ) -> str: + normalized = _validate_wait_for_args( + condition=condition, + text=text, + window_name=window_name, + timeout=timeout, + interval=interval, + ) + desktop = get_desktop() + use_dom_bool = _as_bool(use_dom) + started_at = time.monotonic() + deadline = started_at + timeout + attempts = 0 + last_detail = "condition was not evaluated" + + while True: + attempts += 1 + desktop_state = desktop.get_state( + use_vision=False, + use_dom=use_dom_bool, + use_ui_tree=True, + use_annotation=False, + ) + matched, last_detail = _matches_wait_condition( + desktop_state=desktop_state, + condition=normalized, + text=text, + window_name=window_name, + ) + if matched: + elapsed = time.monotonic() - started_at + return ( + f"WaitFor condition '{normalized}' satisfied after " + f"{elapsed:.2f}s and {attempts} attempt(s): {last_detail}." + ) + + remaining = deadline - time.monotonic() + if remaining <= 0: + raise TimeoutError( + f"Timed out after {timeout:.2f}s waiting for '{normalized}': {last_detail}." + ) + time.sleep(min(interval, remaining)) diff --git a/tests/test_wait_for_tool.py b/tests/test_wait_for_tool.py new file mode 100644 index 00000000..0347c1c5 --- /dev/null +++ b/tests/test_wait_for_tool.py @@ -0,0 +1,201 @@ +from __future__ import annotations + +import asyncio +from collections.abc import Callable + +import pytest + +from windows_mcp.desktop.views import DesktopState, Status, Window +from windows_mcp.tools.input import register +from windows_mcp.tree.views import BoundingBox, Center, TextElementNode, TreeElementNode, TreeState + + +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, states: list[DesktopState]) -> None: + self.states = states + self.desktop_state: DesktopState | None = None + self.calls: list[dict[str, object]] = [] + + def get_state(self, **kwargs: object) -> DesktopState: + self.calls.append(kwargs) + if self.states: + self.desktop_state = self.states.pop(0) + if self.desktop_state is None: + raise RuntimeError("FakeDesktop has no desktop state") + return self.desktop_state + + +def _box() -> BoundingBox: + return BoundingBox(left=0, top=0, right=100, bottom=40, width=100, height=40) + + +def _window(name: str) -> Window: + return Window( + name=name, + is_browser=False, + depth=0, + status=Status.NORMAL, + bounding_box=_box(), + handle=123, + process_id=456, + ) + + +def _element( + name: str, + *, + window_name: str = "Notepad", + focused: bool = False, +) -> TreeElementNode: + return TreeElementNode( + name=name, + control_type="Button", + window_name=window_name, + bounding_box=_box(), + center=Center(x=50, y=20), + metadata={"has_focused": focused}, + ) + + +def _state( + *, + active_window_name: str = "Notepad", + elements: list[TreeElementNode] | None = None, + dom_texts: list[str] | None = None, +) -> DesktopState: + active_window = _window(active_window_name) + return DesktopState( + active_desktop={"name": "Desktop 1"}, + all_desktops=[], + active_window=active_window, + windows=[], + tree_state=TreeState( + interactive_nodes=elements or [], + dom_informative_nodes=[TextElementNode(text=text) for text in (dom_texts or [])], + ), + ) + + +def _register_tools(desktop: FakeDesktop) -> dict[str, Callable]: + mcp = FakeMCP() + register(mcp, get_desktop=lambda: desktop, get_analytics=lambda: None) + return mcp.tools + + +def test_wait_for_text_polls_until_dom_text_appears() -> None: + desktop = FakeDesktop( + [ + _state(dom_texts=["Loading"]), + _state(dom_texts=["Ready for search"]), + ] + ) + tools = _register_tools(desktop) + + result = asyncio.run( + tools["WaitFor"]( + condition="text_exists", + text="ready", + timeout=1, + interval=0.001, + use_dom=True, + ) + ) + + assert "condition 'text_exists' satisfied" in result + assert len(desktop.calls) == 2 + assert desktop.calls[0]["use_dom"] is True + assert desktop.calls[0]["use_vision"] is False + + +def test_wait_for_active_window_matches_by_window_name() -> None: + desktop = FakeDesktop([_state(active_window_name="Untitled - Notepad")]) + tools = _register_tools(desktop) + + result = asyncio.run( + tools["WaitFor"]( + condition="active_window", + window_name="notepad", + timeout=1, + interval=0.001, + ) + ) + + assert "active window matched" in result + + +def test_wait_for_focused_element_matches_text_and_window() -> None: + desktop = FakeDesktop( + [ + _state( + elements=[ + _element("Cancel", focused=False), + _element("Search", window_name="Command Palette", focused=True), + ] + ) + ] + ) + tools = _register_tools(desktop) + + result = asyncio.run( + tools["WaitFor"]( + condition="focused_element", + text="search", + window_name="palette", + timeout=1, + interval=0.001, + ) + ) + + assert "focused element matched 'Search'" in result + + +def test_wait_for_element_enabled_alias_matches_interactive_node() -> None: + desktop = FakeDesktop([_state(elements=[_element("Submit", window_name="Checkout")])]) + tools = _register_tools(desktop) + + result = asyncio.run( + tools["WaitFor"]( + condition="enabled", + text="submit", + window_name="checkout", + timeout=1, + interval=0.001, + ) + ) + + assert "condition 'element_enabled' satisfied" in result + + +def test_wait_for_text_requires_text() -> None: + desktop = FakeDesktop([_state()]) + tools = _register_tools(desktop) + + with pytest.raises(ValueError, match="text is required"): + asyncio.run(tools["WaitFor"](condition="text_exists")) + + +def test_wait_for_timeout_reports_last_observed_state() -> None: + desktop = FakeDesktop([_state(active_window_name="Explorer")]) + tools = _register_tools(desktop) + + with pytest.raises(TimeoutError, match="active window was 'Explorer'"): + asyncio.run( + tools["WaitFor"]( + condition="active_window", + window_name="Terminal", + timeout=0.01, + interval=0.001, + ) + )