mirror of
https://github.com/hanzoai/python-sdk.git
synced 2026-10-23 20:23:00 +00:00
1826 lines
65 KiB
Python
1826 lines
65 KiB
Python
"""Unified computer control tool with native API acceleration.
|
|
|
|
Cross-platform support:
|
|
- macOS: Direct Quartz/CoreGraphics APIs (fastest, ~10-50x faster than pyautogui)
|
|
- Linux: xdotool/scrot for X11
|
|
- Windows: ctypes win32api
|
|
|
|
Performance:
|
|
- Native path: <5ms click, <2ms keypress, <50ms screenshot
|
|
- Fallback to pyautogui when native APIs unavailable
|
|
- Zero-delay operations in native mode
|
|
- Batch operations with minimal overhead
|
|
"""
|
|
|
|
import os
|
|
import re
|
|
import sys
|
|
import json
|
|
import time
|
|
import shutil
|
|
import asyncio
|
|
import tempfile
|
|
import subprocess
|
|
from typing import Any, Literal, Optional, Annotated, final, override
|
|
from pathlib import Path
|
|
from concurrent.futures import ThreadPoolExecutor
|
|
|
|
from pydantic import Field
|
|
from mcp.server import FastMCP
|
|
from mcp.server.fastmcp import Context as MCPContext
|
|
|
|
from hanzo_tools.core import BaseTool, PermissionManager, auto_timeout, capture
|
|
from hanzo_tools.core.unified import _result_to_mcp
|
|
|
|
# Platform detection
|
|
PLATFORM = sys.platform
|
|
IS_MACOS = PLATFORM == "darwin"
|
|
IS_LINUX = PLATFORM.startswith("linux")
|
|
IS_WINDOWS = PLATFORM == "win32"
|
|
|
|
# macOS: Quartz/CoreGraphics
|
|
QUARTZ_AVAILABLE = False
|
|
if IS_MACOS:
|
|
try:
|
|
import AppKit
|
|
import Quartz
|
|
from Quartz import (
|
|
CGEventPost,
|
|
CGMainDisplayID,
|
|
CGDisplayPixelsHigh,
|
|
CGDisplayPixelsWide,
|
|
CGEventCreateMouseEvent,
|
|
CGEventCreateKeyboardEvent,
|
|
CGEventCreateScrollWheelEvent,
|
|
kCGHIDEventTap,
|
|
kCGEventMouseMoved,
|
|
kCGMouseButtonLeft,
|
|
kCGEventLeftMouseUp,
|
|
kCGMouseButtonRight,
|
|
kCGEventOtherMouseUp,
|
|
kCGEventRightMouseUp,
|
|
kCGMouseButtonCenter,
|
|
kCGEventLeftMouseDown,
|
|
kCGEventOtherMouseDown,
|
|
kCGEventRightMouseDown,
|
|
kCGScrollEventUnitLine,
|
|
kCGEventLeftMouseDragged,
|
|
kCGEventOtherMouseDragged,
|
|
kCGEventRightMouseDragged,
|
|
)
|
|
|
|
QUARTZ_AVAILABLE = True
|
|
except ImportError:
|
|
QUARTZ_AVAILABLE = False
|
|
|
|
# Linux: Check for X11 tools
|
|
XDOTOOL_AVAILABLE = False
|
|
SCROT_AVAILABLE = False
|
|
if IS_LINUX:
|
|
XDOTOOL_AVAILABLE = shutil.which("xdotool") is not None
|
|
SCROT_AVAILABLE = shutil.which("scrot") is not None
|
|
|
|
# Windows: ctypes for native API
|
|
WIN32_AVAILABLE = False
|
|
if IS_WINDOWS:
|
|
try:
|
|
import ctypes
|
|
|
|
WIN32_AVAILABLE = True
|
|
except ImportError:
|
|
WIN32_AVAILABLE = False
|
|
|
|
# Check if we have any native backend
|
|
NATIVE_AVAILABLE = QUARTZ_AVAILABLE or XDOTOOL_AVAILABLE or WIN32_AVAILABLE
|
|
|
|
# --- ydotool / uinput backend (Linux) ----------------------------------------
|
|
# xdotool/XTEST synthetic button events CRASH Tauri/WebKitGTK apps with
|
|
# `Gdk-CRITICAL gdk_event_set_source_device: assertion failed`. ydotool injects
|
|
# real /dev/uinput events (via ydotoold) that don't crash and register in the
|
|
# webview. Prefer ydotool over xdotool whenever the daemon socket is present.
|
|
import os as _os
|
|
YDOTOOL_BIN = shutil.which("ydotool")
|
|
_YD_SOCK = "/run/ydotoold.sock"
|
|
if not _os.path.exists(_YD_SOCK):
|
|
_alt = "/run/user/%d/.ydotool_socket" % _os.getuid()
|
|
if _os.path.exists(_alt):
|
|
_YD_SOCK = _alt
|
|
YDOTOOL_AVAILABLE = bool(YDOTOOL_BIN) and _os.path.exists(_YD_SOCK)
|
|
_YD_ENV = dict(_os.environ, YDOTOOL_SOCKET=_YD_SOCK)
|
|
_YD_CLICK = {"left": "0xC0", "right": "0xC1", "middle": "0xC2"}
|
|
_YD_DOWN = {"left": "0x40", "right": "0x41", "middle": "0x42"}
|
|
_YD_UP = {"left": "0x80", "right": "0x81", "middle": "0x82"}
|
|
_YD_KEYS = {
|
|
'a':30,'b':48,'c':46,'d':32,'e':18,'f':33,'g':34,'h':35,'i':23,'j':36,'k':37,
|
|
'l':38,'m':50,'n':49,'o':24,'p':25,'q':16,'r':19,'s':31,'t':20,'u':22,'v':47,
|
|
'w':17,'x':45,'y':21,'z':44,'1':2,'2':3,'3':4,'4':5,'5':6,'6':7,'7':8,'8':9,
|
|
'9':10,'0':11,'enter':28,'return':28,'\n':28,'tab':15,'space':57,' ':57,
|
|
'esc':1,'escape':1,'backspace':14,'delete':111,'del':111,'insert':110,
|
|
'home':102,'end':107,'pageup':104,'pagedown':109,'up':103,'down':108,
|
|
'left':105,'right':106,'ctrl':29,'control':29,'leftctrl':29,'shift':42,
|
|
'leftshift':42,'alt':56,'leftalt':56,'meta':125,'cmd':125,'command':125,
|
|
'win':125,'super':125,'minus':12,'-':12,'equal':13,'=':13,'comma':51,',':51,
|
|
'period':52,'dot':52,'.':52,'slash':53,'/':53,'semicolon':39,';':39,
|
|
'capslock':58,'f1':59,'f2':60,'f3':61,'f4':62,'f5':63,'f6':64,'f7':65,
|
|
'f8':66,'f9':67,'f10':68,'f11':87,'f12':88,
|
|
}
|
|
def _yd(*args, timeout=5):
|
|
subprocess.run([YDOTOOL_BIN, *[str(a) for a in args]],
|
|
capture_output=True, timeout=timeout, env=_YD_ENV)
|
|
def _yd_move(x, y):
|
|
_yd("mousemove", "-a", "-x", int(x), "-y", int(y))
|
|
def _yd_click(x, y, button="left"):
|
|
_yd_move(x, y); time.sleep(0.02); _yd("click", _YD_CLICK.get(button, "0xC0"))
|
|
def _yd_keycode(name):
|
|
return _YD_KEYS.get(str(name).lower())
|
|
def _yd_press(key):
|
|
c = _yd_keycode(key)
|
|
if c is not None: _yd("key", "%d:1" % c, "%d:0" % c)
|
|
def _yd_hotkey(*keys):
|
|
codes = [_yd_keycode(k) for k in keys]
|
|
if any(c is None for c in codes): return
|
|
_yd("key", *["%d:1" % c for c in codes], *["%d:0" % c for c in reversed(codes)])
|
|
def _yd_type(text, interval=0):
|
|
if interval and interval > 0:
|
|
_yd("type", "--key-delay", int(interval * 1000), str(text), timeout=30)
|
|
else:
|
|
_yd("type", str(text), timeout=30)
|
|
# --- end ydotool backend -----------------------------------------------------
|
|
|
|
# Shared thread pool
|
|
_EXECUTOR = ThreadPoolExecutor(max_workers=4, thread_name_prefix="computer_")
|
|
|
|
# macOS key codes
|
|
KEY_CODES = {
|
|
"a": 0x00,
|
|
"s": 0x01,
|
|
"d": 0x02,
|
|
"f": 0x03,
|
|
"h": 0x04,
|
|
"g": 0x05,
|
|
"z": 0x06,
|
|
"x": 0x07,
|
|
"c": 0x08,
|
|
"v": 0x09,
|
|
"b": 0x0B,
|
|
"q": 0x0C,
|
|
"w": 0x0D,
|
|
"e": 0x0E,
|
|
"r": 0x0F,
|
|
"y": 0x10,
|
|
"t": 0x11,
|
|
"1": 0x12,
|
|
"2": 0x13,
|
|
"3": 0x14,
|
|
"4": 0x15,
|
|
"5": 0x17,
|
|
"6": 0x16,
|
|
"7": 0x1A,
|
|
"8": 0x1C,
|
|
"9": 0x19,
|
|
"0": 0x1D,
|
|
"-": 0x1B,
|
|
"=": 0x18,
|
|
"[": 0x21,
|
|
"]": 0x1E,
|
|
"\\": 0x2A,
|
|
";": 0x29,
|
|
"'": 0x27,
|
|
"`": 0x32,
|
|
",": 0x2B,
|
|
".": 0x2F,
|
|
"/": 0x2C,
|
|
"o": 0x1F,
|
|
"u": 0x20,
|
|
"i": 0x22,
|
|
"p": 0x23,
|
|
"l": 0x25,
|
|
"j": 0x26,
|
|
"k": 0x28,
|
|
"n": 0x2D,
|
|
"m": 0x2E,
|
|
" ": 0x31,
|
|
"space": 0x31,
|
|
"return": 0x24,
|
|
"enter": 0x24,
|
|
"\n": 0x24,
|
|
"\r": 0x24,
|
|
"tab": 0x30,
|
|
"\t": 0x30,
|
|
"backspace": 0x33,
|
|
"\b": 0x33,
|
|
"escape": 0x35,
|
|
"esc": 0x35,
|
|
"command": 0x37,
|
|
"cmd": 0x37,
|
|
"shift": 0x38,
|
|
"shiftleft": 0x38,
|
|
"shiftright": 0x3C,
|
|
"capslock": 0x39,
|
|
"option": 0x3A,
|
|
"alt": 0x3A,
|
|
"optionleft": 0x3A,
|
|
"altleft": 0x3A,
|
|
"optionright": 0x3D,
|
|
"altright": 0x3D,
|
|
"control": 0x3B,
|
|
"ctrl": 0x3B,
|
|
"ctrlleft": 0x3B,
|
|
"ctrlright": 0x3E,
|
|
"fn": 0x3F,
|
|
"f1": 0x7A,
|
|
"f2": 0x78,
|
|
"f3": 0x63,
|
|
"f4": 0x76,
|
|
"f5": 0x60,
|
|
"f6": 0x61,
|
|
"f7": 0x62,
|
|
"f8": 0x64,
|
|
"f9": 0x65,
|
|
"f10": 0x6D,
|
|
"f11": 0x67,
|
|
"f12": 0x6F,
|
|
"home": 0x73,
|
|
"end": 0x77,
|
|
"pageup": 0x74,
|
|
"pagedown": 0x79,
|
|
"delete": 0x75,
|
|
"del": 0x75,
|
|
"left": 0x7B,
|
|
"right": 0x7C,
|
|
"down": 0x7D,
|
|
"up": 0x7E,
|
|
}
|
|
|
|
SHIFT_CHARS = '~!@#$%^&*()_+{}|:"<>?ABCDEFGHIJKLMNOPQRSTUVWXYZ'
|
|
|
|
|
|
def _get_vk_code(key: str) -> int | None:
|
|
"""Get Windows virtual key code."""
|
|
VK_CODES = {
|
|
"a": 0x41,
|
|
"b": 0x42,
|
|
"c": 0x43,
|
|
"d": 0x44,
|
|
"e": 0x45,
|
|
"f": 0x46,
|
|
"g": 0x47,
|
|
"h": 0x48,
|
|
"i": 0x49,
|
|
"j": 0x4A,
|
|
"k": 0x4B,
|
|
"l": 0x4C,
|
|
"m": 0x4D,
|
|
"n": 0x4E,
|
|
"o": 0x4F,
|
|
"p": 0x50,
|
|
"q": 0x51,
|
|
"r": 0x52,
|
|
"s": 0x53,
|
|
"t": 0x54,
|
|
"u": 0x55,
|
|
"v": 0x56,
|
|
"w": 0x57,
|
|
"x": 0x58,
|
|
"y": 0x59,
|
|
"z": 0x5A,
|
|
"0": 0x30,
|
|
"1": 0x31,
|
|
"2": 0x32,
|
|
"3": 0x33,
|
|
"4": 0x34,
|
|
"5": 0x35,
|
|
"6": 0x36,
|
|
"7": 0x37,
|
|
"8": 0x38,
|
|
"9": 0x39,
|
|
"return": 0x0D,
|
|
"enter": 0x0D,
|
|
"tab": 0x09,
|
|
"space": 0x20,
|
|
"backspace": 0x08,
|
|
"escape": 0x1B,
|
|
"esc": 0x1B,
|
|
"shift": 0x10,
|
|
"ctrl": 0x11,
|
|
"control": 0x11,
|
|
"alt": 0x12,
|
|
"left": 0x25,
|
|
"up": 0x26,
|
|
"right": 0x27,
|
|
"down": 0x28,
|
|
"delete": 0x2E,
|
|
"home": 0x24,
|
|
"end": 0x23,
|
|
"pageup": 0x21,
|
|
"pagedown": 0x22,
|
|
"f1": 0x70,
|
|
"f2": 0x71,
|
|
"f3": 0x72,
|
|
"f4": 0x73,
|
|
"f5": 0x74,
|
|
"f6": 0x75,
|
|
"f7": 0x76,
|
|
"f8": 0x77,
|
|
"f9": 0x78,
|
|
"f10": 0x79,
|
|
"f11": 0x7A,
|
|
"f12": 0x7B,
|
|
}
|
|
return VK_CODES.get(key.lower())
|
|
|
|
|
|
class NativeControl:
|
|
"""Cross-platform native control - uses fastest available backend."""
|
|
|
|
@staticmethod
|
|
def get_platform_info() -> dict:
|
|
"""Get platform capabilities."""
|
|
return {
|
|
"platform": PLATFORM,
|
|
"native_available": NATIVE_AVAILABLE,
|
|
"backends": {
|
|
"quartz": QUARTZ_AVAILABLE,
|
|
"xdotool": XDOTOOL_AVAILABLE,
|
|
"scrot": SCROT_AVAILABLE,
|
|
"win32": WIN32_AVAILABLE,
|
|
},
|
|
}
|
|
|
|
@staticmethod
|
|
def mouse_position() -> tuple[int, int]:
|
|
"""Get mouse position."""
|
|
if IS_MACOS and QUARTZ_AVAILABLE:
|
|
loc = AppKit.NSEvent.mouseLocation()
|
|
return int(loc.x), int(CGDisplayPixelsHigh(0) - loc.y)
|
|
elif IS_LINUX and XDOTOOL_AVAILABLE:
|
|
result = subprocess.run(
|
|
["xdotool", "getmouselocation", "--shell"],
|
|
capture_output=True,
|
|
text=True,
|
|
timeout=2,
|
|
)
|
|
vals = dict(
|
|
line.split("=")
|
|
for line in result.stdout.strip().split("\n")
|
|
if "=" in line
|
|
)
|
|
return int(vals.get("X", 0)), int(vals.get("Y", 0))
|
|
elif IS_WINDOWS and WIN32_AVAILABLE:
|
|
|
|
class POINT(ctypes.Structure):
|
|
_fields_ = [("x", ctypes.c_long), ("y", ctypes.c_long)]
|
|
|
|
pt = POINT()
|
|
ctypes.windll.user32.GetCursorPos(ctypes.byref(pt))
|
|
return pt.x, pt.y
|
|
else:
|
|
import pyautogui
|
|
|
|
return pyautogui.position()
|
|
|
|
@staticmethod
|
|
def screen_size() -> tuple[int, int]:
|
|
"""Get screen size."""
|
|
if IS_MACOS and QUARTZ_AVAILABLE:
|
|
return CGDisplayPixelsWide(CGMainDisplayID()), CGDisplayPixelsHigh(
|
|
CGMainDisplayID()
|
|
)
|
|
elif IS_LINUX:
|
|
try:
|
|
result = subprocess.run(
|
|
["xdpyinfo"], capture_output=True, text=True, timeout=2
|
|
)
|
|
for line in result.stdout.split("\n"):
|
|
if "dimensions:" in line:
|
|
dims = line.split()[1]
|
|
w, h = dims.split("x")
|
|
return int(w), int(h)
|
|
except Exception:
|
|
pass
|
|
return (1920, 1080)
|
|
elif IS_WINDOWS and WIN32_AVAILABLE:
|
|
return (
|
|
ctypes.windll.user32.GetSystemMetrics(0),
|
|
ctypes.windll.user32.GetSystemMetrics(1),
|
|
)
|
|
else:
|
|
import pyautogui
|
|
|
|
return pyautogui.size()
|
|
|
|
@staticmethod
|
|
def _send_mouse_event_macos(
|
|
event_type: int, x: int, y: int, button: int = 0
|
|
) -> None:
|
|
"""Send mouse event via macOS Quartz."""
|
|
event = CGEventCreateMouseEvent(None, event_type, (x, y), button)
|
|
CGEventPost(kCGHIDEventTap, event)
|
|
|
|
@staticmethod
|
|
def click(x: int, y: int, button: str = "left") -> None:
|
|
"""Click at position - native speed."""
|
|
if IS_LINUX and YDOTOOL_AVAILABLE:
|
|
_yd_click(x, y, button); return
|
|
if IS_MACOS and QUARTZ_AVAILABLE:
|
|
if button == "left":
|
|
NativeControl._send_mouse_event_macos(
|
|
kCGEventLeftMouseDown, x, y, kCGMouseButtonLeft
|
|
)
|
|
NativeControl._send_mouse_event_macos(
|
|
kCGEventLeftMouseUp, x, y, kCGMouseButtonLeft
|
|
)
|
|
elif button == "right":
|
|
NativeControl._send_mouse_event_macos(
|
|
kCGEventRightMouseDown, x, y, kCGMouseButtonRight
|
|
)
|
|
NativeControl._send_mouse_event_macos(
|
|
kCGEventRightMouseUp, x, y, kCGMouseButtonRight
|
|
)
|
|
elif button == "middle":
|
|
NativeControl._send_mouse_event_macos(
|
|
kCGEventOtherMouseDown, x, y, kCGMouseButtonCenter
|
|
)
|
|
NativeControl._send_mouse_event_macos(
|
|
kCGEventOtherMouseUp, x, y, kCGMouseButtonCenter
|
|
)
|
|
elif IS_LINUX and XDOTOOL_AVAILABLE:
|
|
btn_map = {"left": "1", "middle": "2", "right": "3"}
|
|
subprocess.run(
|
|
[
|
|
"xdotool",
|
|
"mousemove",
|
|
str(x),
|
|
str(y),
|
|
"click",
|
|
btn_map.get(button, "1"),
|
|
],
|
|
capture_output=True,
|
|
timeout=2,
|
|
)
|
|
elif IS_WINDOWS and WIN32_AVAILABLE:
|
|
ctypes.windll.user32.SetCursorPos(x, y)
|
|
if button == "left":
|
|
ctypes.windll.user32.mouse_event(0x0002, 0, 0, 0, 0)
|
|
ctypes.windll.user32.mouse_event(0x0004, 0, 0, 0, 0)
|
|
elif button == "right":
|
|
ctypes.windll.user32.mouse_event(0x0008, 0, 0, 0, 0)
|
|
ctypes.windll.user32.mouse_event(0x0010, 0, 0, 0, 0)
|
|
elif button == "middle":
|
|
ctypes.windll.user32.mouse_event(0x0020, 0, 0, 0, 0)
|
|
ctypes.windll.user32.mouse_event(0x0040, 0, 0, 0, 0)
|
|
else:
|
|
import pyautogui
|
|
|
|
pyautogui.click(x, y, button=button)
|
|
|
|
@staticmethod
|
|
def double_click(x: int, y: int) -> None:
|
|
"""Double click."""
|
|
if IS_LINUX and YDOTOOL_AVAILABLE:
|
|
_yd_move(x, y); _yd("click", "0xC0"); _yd("click", "0xC0"); return
|
|
if IS_LINUX and XDOTOOL_AVAILABLE:
|
|
subprocess.run(
|
|
["xdotool", "mousemove", str(x), str(y), "click", "--repeat", "2", "1"],
|
|
capture_output=True,
|
|
timeout=2,
|
|
)
|
|
else:
|
|
NativeControl.click(x, y)
|
|
time.sleep(0.01)
|
|
NativeControl.click(x, y)
|
|
|
|
@staticmethod
|
|
def move(x: int, y: int) -> None:
|
|
"""Move mouse."""
|
|
if IS_LINUX and YDOTOOL_AVAILABLE:
|
|
_yd_move(x, y); return
|
|
if IS_MACOS and QUARTZ_AVAILABLE:
|
|
NativeControl._send_mouse_event_macos(kCGEventMouseMoved, x, y, 0)
|
|
elif IS_LINUX and XDOTOOL_AVAILABLE:
|
|
subprocess.run(
|
|
["xdotool", "mousemove", str(x), str(y)], capture_output=True, timeout=2
|
|
)
|
|
elif IS_WINDOWS and WIN32_AVAILABLE:
|
|
ctypes.windll.user32.SetCursorPos(x, y)
|
|
else:
|
|
import pyautogui
|
|
|
|
pyautogui.moveTo(x, y, _pause=False)
|
|
|
|
@staticmethod
|
|
def drag(
|
|
start_x: int, start_y: int, end_x: int, end_y: int, button: str = "left"
|
|
) -> None:
|
|
"""Drag from start to end."""
|
|
if IS_LINUX and YDOTOOL_AVAILABLE:
|
|
_yd_move(start_x, start_y); _yd("click", _YD_DOWN.get(button, "0x40"))
|
|
_yd_move(end_x, end_y); _yd("click", _YD_UP.get(button, "0x80")); return
|
|
if IS_MACOS and QUARTZ_AVAILABLE:
|
|
if button == "left":
|
|
NativeControl._send_mouse_event_macos(
|
|
kCGEventLeftMouseDown, start_x, start_y, kCGMouseButtonLeft
|
|
)
|
|
drag_type = kCGEventLeftMouseDragged
|
|
up_type = kCGEventLeftMouseUp
|
|
btn = kCGMouseButtonLeft
|
|
elif button == "right":
|
|
NativeControl._send_mouse_event_macos(
|
|
kCGEventRightMouseDown, start_x, start_y, kCGMouseButtonRight
|
|
)
|
|
drag_type = kCGEventRightMouseDragged
|
|
up_type = kCGEventRightMouseUp
|
|
btn = kCGMouseButtonRight
|
|
else:
|
|
NativeControl._send_mouse_event_macos(
|
|
kCGEventOtherMouseDown, start_x, start_y, kCGMouseButtonCenter
|
|
)
|
|
drag_type = kCGEventOtherMouseDragged
|
|
up_type = kCGEventOtherMouseUp
|
|
btn = kCGMouseButtonCenter
|
|
|
|
steps = max(abs(end_x - start_x), abs(end_y - start_y)) // 10 or 1
|
|
for i in range(1, steps + 1):
|
|
cx = start_x + (end_x - start_x) * i // steps
|
|
cy = start_y + (end_y - start_y) * i // steps
|
|
NativeControl._send_mouse_event_macos(drag_type, cx, cy, btn)
|
|
time.sleep(0.001)
|
|
NativeControl._send_mouse_event_macos(up_type, end_x, end_y, btn)
|
|
elif IS_LINUX and XDOTOOL_AVAILABLE:
|
|
btn_map = {"left": "1", "middle": "2", "right": "3"}
|
|
subprocess.run(
|
|
[
|
|
"xdotool",
|
|
"mousemove",
|
|
str(start_x),
|
|
str(start_y),
|
|
"mousedown",
|
|
btn_map.get(button, "1"),
|
|
"mousemove",
|
|
str(end_x),
|
|
str(end_y),
|
|
"mouseup",
|
|
btn_map.get(button, "1"),
|
|
],
|
|
capture_output=True,
|
|
timeout=5,
|
|
)
|
|
elif IS_WINDOWS and WIN32_AVAILABLE:
|
|
ctypes.windll.user32.SetCursorPos(start_x, start_y)
|
|
ctypes.windll.user32.mouse_event(0x0002, 0, 0, 0, 0)
|
|
steps = max(abs(end_x - start_x), abs(end_y - start_y)) // 10 or 1
|
|
for i in range(1, steps + 1):
|
|
cx = start_x + (end_x - start_x) * i // steps
|
|
cy = start_y + (end_y - start_y) * i // steps
|
|
ctypes.windll.user32.SetCursorPos(cx, cy)
|
|
time.sleep(0.001)
|
|
ctypes.windll.user32.mouse_event(0x0004, 0, 0, 0, 0)
|
|
else:
|
|
import pyautogui
|
|
|
|
pyautogui.moveTo(start_x, start_y, _pause=False)
|
|
pyautogui.drag(end_x - start_x, end_y - start_y, _pause=False)
|
|
|
|
@staticmethod
|
|
def scroll(amount: int, x: int | None = None, y: int | None = None) -> None:
|
|
"""Scroll."""
|
|
if x is not None and y is not None:
|
|
NativeControl.move(x, y)
|
|
|
|
if IS_MACOS and QUARTZ_AVAILABLE:
|
|
event = CGEventCreateScrollWheelEvent(
|
|
None, kCGScrollEventUnitLine, 1, amount
|
|
)
|
|
CGEventPost(kCGHIDEventTap, event)
|
|
elif IS_LINUX and XDOTOOL_AVAILABLE:
|
|
btn = "4" if amount > 0 else "5"
|
|
for _ in range(abs(amount)):
|
|
subprocess.run(
|
|
["xdotool", "click", btn], capture_output=True, timeout=2
|
|
)
|
|
elif IS_WINDOWS and WIN32_AVAILABLE:
|
|
ctypes.windll.user32.mouse_event(0x0800, 0, 0, amount * 120, 0)
|
|
else:
|
|
import pyautogui
|
|
|
|
pyautogui.scroll(amount, _pause=False)
|
|
|
|
@staticmethod
|
|
def _send_key_event_macos(key_code: int, down: bool) -> None:
|
|
"""Send key event via macOS Quartz."""
|
|
event = CGEventCreateKeyboardEvent(None, key_code, down)
|
|
CGEventPost(kCGHIDEventTap, event)
|
|
|
|
@staticmethod
|
|
def key_down(key: str) -> None:
|
|
"""Press key down."""
|
|
if IS_LINUX and YDOTOOL_AVAILABLE:
|
|
_c = _yd_keycode(key)
|
|
if _c is not None: _yd("key", "%d:1" % _c)
|
|
return
|
|
key_lower = key.lower()
|
|
if IS_MACOS and QUARTZ_AVAILABLE:
|
|
if key_lower in KEY_CODES:
|
|
NativeControl._send_key_event_macos(KEY_CODES[key_lower], True)
|
|
elif IS_LINUX and XDOTOOL_AVAILABLE:
|
|
subprocess.run(
|
|
["xdotool", "keydown", key_lower], capture_output=True, timeout=2
|
|
)
|
|
elif IS_WINDOWS and WIN32_AVAILABLE:
|
|
vk = _get_vk_code(key_lower)
|
|
if vk:
|
|
ctypes.windll.user32.keybd_event(vk, 0, 0, 0)
|
|
else:
|
|
import pyautogui
|
|
|
|
pyautogui.keyDown(key_lower, _pause=False)
|
|
|
|
@staticmethod
|
|
def key_up(key: str) -> None:
|
|
"""Release key."""
|
|
if IS_LINUX and YDOTOOL_AVAILABLE:
|
|
_c = _yd_keycode(key)
|
|
if _c is not None: _yd("key", "%d:0" % _c)
|
|
return
|
|
key_lower = key.lower()
|
|
if IS_MACOS and QUARTZ_AVAILABLE:
|
|
if key_lower in KEY_CODES:
|
|
NativeControl._send_key_event_macos(KEY_CODES[key_lower], False)
|
|
elif IS_LINUX and XDOTOOL_AVAILABLE:
|
|
subprocess.run(
|
|
["xdotool", "keyup", key_lower], capture_output=True, timeout=2
|
|
)
|
|
elif IS_WINDOWS and WIN32_AVAILABLE:
|
|
vk = _get_vk_code(key_lower)
|
|
if vk:
|
|
ctypes.windll.user32.keybd_event(vk, 0, 0x0002, 0)
|
|
else:
|
|
import pyautogui
|
|
|
|
pyautogui.keyUp(key_lower, _pause=False)
|
|
|
|
@staticmethod
|
|
def press(key: str) -> None:
|
|
"""Press and release key."""
|
|
if IS_LINUX and YDOTOOL_AVAILABLE:
|
|
_yd_press(key); return
|
|
if IS_LINUX and XDOTOOL_AVAILABLE:
|
|
subprocess.run(
|
|
["xdotool", "key", key.lower()], capture_output=True, timeout=2
|
|
)
|
|
else:
|
|
NativeControl.key_down(key)
|
|
NativeControl.key_up(key)
|
|
|
|
@staticmethod
|
|
def hotkey(*keys: str) -> None:
|
|
"""Press key combination."""
|
|
if IS_LINUX and YDOTOOL_AVAILABLE:
|
|
_yd_hotkey(*keys); return
|
|
if IS_LINUX and XDOTOOL_AVAILABLE:
|
|
combo = "+".join(keys)
|
|
subprocess.run(["xdotool", "key", combo], capture_output=True, timeout=2)
|
|
else:
|
|
for key in keys:
|
|
NativeControl.key_down(key)
|
|
for key in reversed(keys):
|
|
NativeControl.key_up(key)
|
|
|
|
@staticmethod
|
|
def type_char(char: str) -> None:
|
|
"""Type a single character."""
|
|
if IS_LINUX and YDOTOOL_AVAILABLE:
|
|
_yd_type(char); return
|
|
if IS_LINUX and XDOTOOL_AVAILABLE:
|
|
subprocess.run(
|
|
["xdotool", "type", "--", char], capture_output=True, timeout=2
|
|
)
|
|
elif IS_MACOS and QUARTZ_AVAILABLE:
|
|
if char in SHIFT_CHARS:
|
|
NativeControl.key_down("shift")
|
|
key = char.lower() if char.isalpha() else char
|
|
if key in KEY_CODES:
|
|
NativeControl.press(key)
|
|
NativeControl.key_up("shift")
|
|
elif char.lower() in KEY_CODES:
|
|
NativeControl.press(char.lower())
|
|
else:
|
|
import pyautogui
|
|
|
|
pyautogui.typewrite(char, _pause=False)
|
|
|
|
@staticmethod
|
|
def type_text(text: str, interval: float = 0) -> None:
|
|
"""Type text."""
|
|
if IS_LINUX and YDOTOOL_AVAILABLE:
|
|
_yd_type(text, interval); return
|
|
if IS_LINUX and XDOTOOL_AVAILABLE:
|
|
if interval > 0:
|
|
subprocess.run(
|
|
[
|
|
"xdotool",
|
|
"type",
|
|
"--delay",
|
|
str(int(interval * 1000)),
|
|
"--",
|
|
text,
|
|
],
|
|
capture_output=True,
|
|
timeout=30,
|
|
)
|
|
else:
|
|
subprocess.run(
|
|
["xdotool", "type", "--", text], capture_output=True, timeout=30
|
|
)
|
|
else:
|
|
for char in text:
|
|
NativeControl.type_char(char)
|
|
if interval > 0:
|
|
time.sleep(interval)
|
|
|
|
@staticmethod
|
|
def screenshot_native(region: list[int] | None = None) -> bytes:
|
|
"""Screenshot using native tools - fastest method."""
|
|
with tempfile.NamedTemporaryFile(suffix=".png", delete=False) as f:
|
|
tmp_path = f.name
|
|
|
|
try:
|
|
if IS_MACOS:
|
|
cmd = ["screencapture", "-x", "-t", "png"]
|
|
if region and len(region) == 4:
|
|
x, y, w, h = region
|
|
cmd.extend(["-R", f"{x},{y},{w},{h}"])
|
|
cmd.append(tmp_path)
|
|
subprocess.run(cmd, capture_output=True, timeout=5)
|
|
|
|
elif IS_LINUX and SCROT_AVAILABLE:
|
|
cmd = ["scrot", "-o", tmp_path]
|
|
if region and len(region) == 4:
|
|
x, y, w, h = region
|
|
cmd = ["scrot", "-a", f"{x},{y},{w},{h}", "-o", tmp_path]
|
|
subprocess.run(cmd, capture_output=True, timeout=5)
|
|
|
|
elif IS_WINDOWS or IS_LINUX:
|
|
from PIL import ImageGrab
|
|
|
|
if region and len(region) == 4:
|
|
x, y, w, h = region
|
|
img = ImageGrab.grab(bbox=(x, y, x + w, y + h))
|
|
else:
|
|
img = ImageGrab.grab()
|
|
img.save(tmp_path, "PNG")
|
|
|
|
if os.path.exists(tmp_path) and os.path.getsize(tmp_path) > 0:
|
|
with open(tmp_path, "rb") as f:
|
|
return f.read()
|
|
return b""
|
|
finally:
|
|
if os.path.exists(tmp_path):
|
|
os.unlink(tmp_path)
|
|
|
|
@staticmethod
|
|
def focus_window(title: str) -> bool:
|
|
"""Focus window by app/window name. Supports partial matching on macOS."""
|
|
if IS_MACOS:
|
|
# First try direct app activation
|
|
script = f'tell application "{title}" to activate'
|
|
try:
|
|
result = subprocess.run(
|
|
["osascript", "-e", script], capture_output=True, timeout=10
|
|
)
|
|
if result.returncode == 0:
|
|
return True
|
|
except subprocess.TimeoutExpired:
|
|
pass
|
|
|
|
# If that fails, try to find app by partial name match
|
|
search_script = f"""
|
|
tell application "System Events"
|
|
set matchingApps to (application processes whose name contains "{title}")
|
|
if (count of matchingApps) > 0 then
|
|
set frontApp to item 1 of matchingApps
|
|
set frontmost of frontApp to true
|
|
return name of frontApp
|
|
end if
|
|
end tell
|
|
return ""
|
|
"""
|
|
try:
|
|
result = subprocess.run(
|
|
["osascript", "-e", search_script],
|
|
capture_output=True,
|
|
text=True,
|
|
timeout=10,
|
|
)
|
|
if result.returncode == 0 and result.stdout.strip():
|
|
return True
|
|
except subprocess.TimeoutExpired:
|
|
pass
|
|
return False
|
|
elif IS_LINUX and XDOTOOL_AVAILABLE:
|
|
result = subprocess.run(
|
|
["xdotool", "search", "--name", title, "windowactivate"],
|
|
capture_output=True,
|
|
timeout=10,
|
|
)
|
|
return result.returncode == 0
|
|
elif IS_WINDOWS and WIN32_AVAILABLE:
|
|
hwnd = ctypes.windll.user32.FindWindowW(None, title)
|
|
if hwnd:
|
|
ctypes.windll.user32.SetForegroundWindow(hwnd)
|
|
return True
|
|
return False
|
|
return False
|
|
|
|
@staticmethod
|
|
def get_active_window() -> dict:
|
|
"""Get active window info."""
|
|
if IS_MACOS:
|
|
script = """
|
|
tell application "System Events"
|
|
set frontApp to first application process whose frontmost is true
|
|
set appName to name of frontApp
|
|
try
|
|
set frontWindow to front window of frontApp
|
|
set winName to name of frontWindow
|
|
set winPos to position of frontWindow
|
|
set winSize to size of frontWindow
|
|
return appName & "|" & winName & "|" & (item 1 of winPos) & "|" & (item 2 of winPos) & "|" & (item 1 of winSize) & "|" & (item 2 of winSize)
|
|
on error
|
|
return appName & "|" & "" & "|0|0|0|0"
|
|
end try
|
|
end tell
|
|
"""
|
|
result = subprocess.run(
|
|
["osascript", "-e", script], capture_output=True, text=True, timeout=5
|
|
)
|
|
if result.returncode == 0:
|
|
parts = result.stdout.strip().split("|")
|
|
if len(parts) >= 6:
|
|
return {
|
|
"app": parts[0],
|
|
"title": parts[1],
|
|
"x": int(parts[2]),
|
|
"y": int(parts[3]),
|
|
"width": int(parts[4]),
|
|
"height": int(parts[5]),
|
|
}
|
|
elif IS_LINUX and XDOTOOL_AVAILABLE:
|
|
result = subprocess.run(
|
|
["xdotool", "getactivewindow", "getwindowname"],
|
|
capture_output=True,
|
|
text=True,
|
|
timeout=5,
|
|
)
|
|
if result.returncode == 0:
|
|
return {"title": result.stdout.strip()}
|
|
elif IS_WINDOWS and WIN32_AVAILABLE:
|
|
hwnd = ctypes.windll.user32.GetForegroundWindow()
|
|
length = ctypes.windll.user32.GetWindowTextLengthW(hwnd)
|
|
buf = ctypes.create_unicode_buffer(length + 1)
|
|
ctypes.windll.user32.GetWindowTextW(hwnd, buf, length + 1)
|
|
return {"title": buf.value, "hwnd": hwnd}
|
|
return {"error": "Could not get active window"}
|
|
|
|
@staticmethod
|
|
def list_windows() -> list[dict]:
|
|
"""List all windows."""
|
|
if IS_MACOS:
|
|
# Use \x1f (unit separator) as delimiter - unlikely to appear in window titles
|
|
script = """
|
|
set windowList to ""
|
|
set delim to ASCII character 31
|
|
tell application "System Events"
|
|
set allProcesses to application processes whose visible is true
|
|
repeat with proc in allProcesses
|
|
set procName to name of proc
|
|
try
|
|
set procWindows to windows of proc
|
|
repeat with win in procWindows
|
|
set winName to name of win
|
|
set winPos to position of win
|
|
set winSize to size of win
|
|
set windowList to windowList & procName & delim & winName & delim & (item 1 of winPos) & delim & (item 2 of winPos) & delim & (item 1 of winSize) & delim & (item 2 of winSize) & "\\n"
|
|
end repeat
|
|
end try
|
|
end repeat
|
|
end tell
|
|
return windowList
|
|
"""
|
|
result = subprocess.run(
|
|
["osascript", "-e", script], capture_output=True, text=True, timeout=10
|
|
)
|
|
if result.returncode == 0:
|
|
windows = []
|
|
for line in result.stdout.strip().split("\n"):
|
|
if "\x1f" in line:
|
|
parts = line.split("\x1f")
|
|
if len(parts) >= 6:
|
|
try:
|
|
windows.append(
|
|
{
|
|
"app": parts[0],
|
|
"title": parts[1],
|
|
"x": int(parts[2]),
|
|
"y": int(parts[3]),
|
|
"width": int(parts[4]),
|
|
"height": int(parts[5]),
|
|
}
|
|
)
|
|
except (ValueError, IndexError):
|
|
# Skip windows with parsing errors
|
|
continue
|
|
return windows
|
|
return []
|
|
|
|
|
|
Action = Literal[
|
|
# Mouse
|
|
"click",
|
|
"double_click",
|
|
"right_click",
|
|
"middle_click",
|
|
"move",
|
|
"move_relative",
|
|
"drag",
|
|
"drag_relative",
|
|
"scroll",
|
|
# Touch (mobile emulation)
|
|
"tap",
|
|
"swipe",
|
|
"pinch",
|
|
# Keyboard
|
|
"type",
|
|
"write",
|
|
"press",
|
|
"key_down",
|
|
"key_up",
|
|
"hotkey",
|
|
# Screen capture
|
|
"screenshot",
|
|
"screenshot_region",
|
|
"capture",
|
|
# Screen recording (consolidated from ScreenTool)
|
|
"session", # ONE-SHOT: record → analyze → compress → return for Claude
|
|
"record", # Start background recording
|
|
"stop", # Stop recording and get compressed frames
|
|
"analyze", # Analyze existing video file
|
|
# Image location
|
|
"locate",
|
|
"locate_all",
|
|
"locate_center",
|
|
"wait_for_image",
|
|
"wait_while_image",
|
|
# Pixel
|
|
"pixel",
|
|
"pixel_matches",
|
|
# Window
|
|
"get_active_window",
|
|
"list_windows",
|
|
"focus_window",
|
|
# Screen info
|
|
"get_screens",
|
|
"screen_size",
|
|
"current_screen",
|
|
# Region helpers
|
|
"define_region",
|
|
"region_screenshot",
|
|
"region_locate",
|
|
# Timing
|
|
"sleep",
|
|
"countdown",
|
|
"set_pause",
|
|
"set_failsafe",
|
|
# Batch
|
|
"batch",
|
|
# Info
|
|
"info",
|
|
"position",
|
|
"status",
|
|
]
|
|
|
|
|
|
@final
|
|
class ComputerTool(BaseTool):
|
|
"""Unified computer control with native API acceleration.
|
|
|
|
Cross-platform support:
|
|
- macOS: Quartz/CoreGraphics APIs (~10-50x faster)
|
|
- Linux: xdotool/scrot for X11
|
|
- Windows: win32api via ctypes
|
|
- Fallback: pyautogui when native unavailable
|
|
|
|
Performance (native mode):
|
|
- Click: <5ms
|
|
- Keypress: <2ms
|
|
- Screenshot: <50ms
|
|
"""
|
|
|
|
name = "computer"
|
|
|
|
def __init__(self, permission_manager: Optional[PermissionManager] = None):
|
|
if permission_manager is None:
|
|
permission_manager = PermissionManager()
|
|
self.permission_manager = permission_manager
|
|
self._pyautogui = None
|
|
self._pil = None
|
|
self._defined_regions: dict[str, tuple[int, int, int, int]] = {}
|
|
self._pause = 0.1
|
|
self._failsafe = True
|
|
|
|
def _ensure_pyautogui(self):
|
|
"""Lazy load pyautogui for fallback/advanced features."""
|
|
if self._pyautogui is None:
|
|
import pyautogui
|
|
|
|
pyautogui.FAILSAFE = self._failsafe
|
|
pyautogui.PAUSE = self._pause
|
|
self._pyautogui = pyautogui
|
|
return self._pyautogui
|
|
|
|
def _ensure_pil(self):
|
|
"""Lazy load PIL."""
|
|
if self._pil is None:
|
|
from PIL import Image
|
|
|
|
self._pil = Image
|
|
return self._pil
|
|
|
|
@property
|
|
@override
|
|
def description(self) -> str:
|
|
platform_info = NativeControl.get_platform_info()
|
|
backends = (
|
|
", ".join(k for k, v in platform_info["backends"].items() if v)
|
|
or "pyautogui"
|
|
)
|
|
return f"""Control local computer with native API acceleration.
|
|
|
|
PLATFORM: {platform_info["platform"]}
|
|
BACKENDS: {backends}
|
|
|
|
MOUSE (< 5ms native):
|
|
- click(x, y) / double_click / right_click / middle_click
|
|
- move(x, y) / move_relative(dx, dy)
|
|
- drag(x, y) / drag_relative(dx, dy)
|
|
- scroll(amount, x, y)
|
|
|
|
KEYBOARD (< 2ms native):
|
|
- type(text, interval): Type text
|
|
- write(text, clear): Type with optional clear
|
|
- press(key): Press and release key
|
|
- key_down(key) / key_up(key): Hold/release
|
|
- hotkey(keys): Key combination ["command", "c"]
|
|
|
|
SCREEN (< 50ms native):
|
|
- screenshot() / screenshot_region(region)
|
|
- get_screens(): List displays
|
|
- screen_size() / current_screen()
|
|
|
|
Screenshots are downscaled JPEG (~1280px, q70) by default to save context; the
|
|
full-resolution capture is saved to a file whose path is returned — pass
|
|
full_res=true (or max_width/quality) only when you need pixel detail.
|
|
|
|
IMAGE LOCATION:
|
|
- locate(image_path): Find image, return center
|
|
- locate_all(image_path): Find all matches
|
|
- wait_for_image(image_path, timeout)
|
|
- wait_while_image(image_path, timeout)
|
|
|
|
PIXEL:
|
|
- pixel(x, y): Get color at point
|
|
- pixel_matches(x, y, color, tolerance)
|
|
|
|
WINDOWS:
|
|
- get_active_window(): Frontmost window info
|
|
- list_windows(): All windows with bounds
|
|
- focus_window(title): Activate window
|
|
|
|
REGIONS:
|
|
- define_region(name, x, y, w, h): Name a region
|
|
- region_screenshot(name): Screenshot region
|
|
- region_locate(name, image): Find in region
|
|
|
|
TIMING:
|
|
- sleep(seconds) / countdown(seconds)
|
|
- set_pause(seconds) / set_failsafe(enabled)
|
|
|
|
BATCH:
|
|
- batch(actions): Execute multiple actions
|
|
|
|
INFO:
|
|
- info() / position()
|
|
|
|
Examples:
|
|
computer(action="click", x=100, y=200)
|
|
computer(action="type", text="Hello")
|
|
computer(action="hotkey", keys=["command", "c"])
|
|
computer(action="screenshot")
|
|
computer(action="batch", actions=[
|
|
{{"action": "click", "x": 100, "y": 200}},
|
|
{{"action": "type", "text": "test"}}
|
|
])
|
|
"""
|
|
|
|
@override
|
|
@auto_timeout("computer")
|
|
async def call(
|
|
self,
|
|
ctx: MCPContext,
|
|
action: str = "info",
|
|
# Coordinates
|
|
x: int | None = None,
|
|
y: int | None = None,
|
|
dx: int | None = None,
|
|
dy: int | None = None,
|
|
end_x: int | None = None,
|
|
end_y: int | None = None,
|
|
# Text/keys
|
|
text: str | None = None,
|
|
key: str | None = None,
|
|
keys: list[str] | None = None,
|
|
# Options
|
|
button: str = "left",
|
|
amount: int | None = None,
|
|
duration: float = 0.25,
|
|
interval: float = 0.02,
|
|
region: list[int] | None = None,
|
|
clear: bool = False,
|
|
# Image location
|
|
image_path: str | None = None,
|
|
confidence: float = 0.9,
|
|
timeout: float = 10.0,
|
|
# Pixel matching
|
|
color: tuple[int, int, int] | list[int] | None = None,
|
|
tolerance: int = 0,
|
|
# Window
|
|
title: str | None = None,
|
|
use_regex: bool = False,
|
|
# Regions
|
|
name: str | None = None,
|
|
width: int | None = None,
|
|
height: int | None = None,
|
|
# Screenshot detail
|
|
max_width: int = 1280,
|
|
quality: int = 70,
|
|
full_res: bool = False,
|
|
# Settings
|
|
value: float | bool | None = None,
|
|
# Batch
|
|
actions: list[dict[str, Any]] | None = None,
|
|
**kwargs,
|
|
) -> Any:
|
|
"""Execute computer control action."""
|
|
loop = asyncio.get_event_loop()
|
|
|
|
def run(fn, *args):
|
|
return loop.run_in_executor(_EXECUTOR, fn, *args)
|
|
|
|
try:
|
|
# Mouse actions - use native when available
|
|
if action == "click":
|
|
if x is None or y is None:
|
|
return json.dumps({"error": "x and y required"})
|
|
await run(NativeControl.click, x, y, button)
|
|
return json.dumps(
|
|
{"success": True, "clicked": [x, y], "button": button}
|
|
)
|
|
|
|
elif action == "double_click":
|
|
if x is None or y is None:
|
|
return json.dumps({"error": "x and y required"})
|
|
await run(NativeControl.double_click, x, y)
|
|
return json.dumps({"success": True, "double_clicked": [x, y]})
|
|
|
|
elif action == "right_click":
|
|
if x is None or y is None:
|
|
return json.dumps({"error": "x and y required"})
|
|
await run(NativeControl.click, x, y, "right")
|
|
return json.dumps({"success": True, "right_clicked": [x, y]})
|
|
|
|
elif action == "middle_click":
|
|
if x is None or y is None:
|
|
return json.dumps({"error": "x and y required"})
|
|
await run(NativeControl.click, x, y, "middle")
|
|
return json.dumps({"success": True, "middle_clicked": [x, y]})
|
|
|
|
elif action == "move":
|
|
if x is None or y is None:
|
|
return json.dumps({"error": "x and y required"})
|
|
await run(NativeControl.move, x, y)
|
|
return json.dumps({"success": True, "moved_to": [x, y]})
|
|
|
|
elif action == "move_relative":
|
|
if dx is None or dy is None:
|
|
return json.dumps({"error": "dx and dy required"})
|
|
pos = NativeControl.mouse_position()
|
|
await run(NativeControl.move, pos[0] + dx, pos[1] + dy)
|
|
return json.dumps({"success": True, "moved_by": [dx, dy]})
|
|
|
|
elif action == "drag":
|
|
if x is None or y is None:
|
|
return json.dumps({"error": "x and y required for drag target"})
|
|
start = NativeControl.mouse_position()
|
|
target_x = end_x if end_x is not None else x
|
|
target_y = end_y if end_y is not None else y
|
|
await run(
|
|
NativeControl.drag, start[0], start[1], target_x, target_y, button
|
|
)
|
|
return json.dumps({"success": True, "dragged_to": [target_x, target_y]})
|
|
|
|
elif action == "drag_relative":
|
|
if dx is None or dy is None:
|
|
return json.dumps({"error": "dx and dy required"})
|
|
pos = NativeControl.mouse_position()
|
|
await run(
|
|
NativeControl.drag, pos[0], pos[1], pos[0] + dx, pos[1] + dy, button
|
|
)
|
|
return json.dumps({"success": True, "dragged_by": [dx, dy]})
|
|
|
|
elif action == "scroll":
|
|
if amount is None:
|
|
return json.dumps({"error": "amount required"})
|
|
await run(NativeControl.scroll, amount, x, y)
|
|
return json.dumps({"success": True, "scrolled": amount})
|
|
|
|
# Keyboard actions
|
|
elif action == "type":
|
|
if not text:
|
|
return json.dumps({"error": "text required"})
|
|
await run(NativeControl.type_text, text, interval)
|
|
return json.dumps({"success": True, "typed": len(text)})
|
|
|
|
elif action == "write":
|
|
if not text:
|
|
return json.dumps({"error": "text required"})
|
|
if clear:
|
|
await run(NativeControl.hotkey, "command", "a")
|
|
await asyncio.sleep(0.05)
|
|
await run(NativeControl.type_text, text, interval)
|
|
return json.dumps(
|
|
{"success": True, "wrote": len(text), "cleared": clear}
|
|
)
|
|
|
|
elif action == "press":
|
|
if not key:
|
|
return json.dumps({"error": "key required"})
|
|
await run(NativeControl.press, key)
|
|
return json.dumps({"success": True, "pressed": key})
|
|
|
|
elif action == "key_down":
|
|
if not key:
|
|
return json.dumps({"error": "key required"})
|
|
await run(NativeControl.key_down, key)
|
|
return json.dumps({"success": True, "key_down": key})
|
|
|
|
elif action == "key_up":
|
|
if not key:
|
|
return json.dumps({"error": "key required"})
|
|
await run(NativeControl.key_up, key)
|
|
return json.dumps({"success": True, "key_up": key})
|
|
|
|
elif action == "hotkey":
|
|
if not keys:
|
|
return json.dumps({"error": "keys required"})
|
|
await run(NativeControl.hotkey, *keys)
|
|
return json.dumps({"success": True, "hotkey": "+".join(keys)})
|
|
|
|
# Screen actions
|
|
elif action == "screenshot" or action == "screenshot_region":
|
|
data = await run(NativeControl.screenshot_native, region)
|
|
return _result_to_mcp(
|
|
capture(
|
|
data,
|
|
fmt="png",
|
|
alt="screen",
|
|
path=self._destination(name),
|
|
max_width=max_width,
|
|
quality=quality,
|
|
full_res=full_res,
|
|
)
|
|
)
|
|
|
|
# Image location (uses pyautogui)
|
|
elif action == "locate":
|
|
if not image_path and not text:
|
|
return json.dumps({"error": "image_path required"})
|
|
path = image_path or text
|
|
result = await run(self._locate, path, confidence, None)
|
|
return result
|
|
|
|
elif action == "locate_all":
|
|
if not image_path and not text:
|
|
return json.dumps({"error": "image_path required"})
|
|
path = image_path or text
|
|
result = await run(self._locate_all, path, confidence)
|
|
return result
|
|
|
|
elif action == "locate_center":
|
|
if not image_path and not text:
|
|
return json.dumps({"error": "image_path required"})
|
|
path = image_path or text
|
|
result = await run(self._locate_center, path, confidence)
|
|
return result
|
|
|
|
elif action == "wait_for_image":
|
|
if not image_path and not text:
|
|
return json.dumps({"error": "image_path required"})
|
|
path = image_path or text
|
|
return await self._wait_for_image(path, timeout, confidence, run)
|
|
|
|
elif action == "wait_while_image":
|
|
if not image_path and not text:
|
|
return json.dumps({"error": "image_path required"})
|
|
path = image_path or text
|
|
return await self._wait_while_image(path, timeout, confidence, run)
|
|
|
|
# Pixel operations
|
|
elif action == "pixel":
|
|
if x is None or y is None:
|
|
return json.dumps({"error": "x and y required"})
|
|
result = await run(self._get_pixel, x, y)
|
|
return result
|
|
|
|
elif action == "pixel_matches":
|
|
if x is None or y is None:
|
|
return json.dumps({"error": "x and y required"})
|
|
if color is None:
|
|
return json.dumps({"error": "color required"})
|
|
result = await run(self._pixel_matches, x, y, tuple(color), tolerance)
|
|
return result
|
|
|
|
# Window management
|
|
elif action == "get_active_window":
|
|
result = await run(NativeControl.get_active_window)
|
|
return json.dumps(result)
|
|
|
|
elif action == "list_windows":
|
|
result = await run(NativeControl.list_windows)
|
|
return json.dumps({"windows": result, "count": len(result)})
|
|
|
|
elif action == "focus_window":
|
|
if not title and not text:
|
|
return json.dumps({"error": "title required"})
|
|
win_title = title or text
|
|
success = await run(NativeControl.focus_window, win_title)
|
|
return json.dumps({"success": success, "focused": win_title})
|
|
|
|
# Screen info
|
|
elif action == "get_screens":
|
|
result = await run(self._get_screens)
|
|
return result
|
|
|
|
elif action == "screen_size":
|
|
size = NativeControl.screen_size()
|
|
return json.dumps({"width": size[0], "height": size[1]})
|
|
|
|
elif action == "current_screen":
|
|
pos = NativeControl.mouse_position()
|
|
size = NativeControl.screen_size()
|
|
return json.dumps(
|
|
{
|
|
"size": {"width": size[0], "height": size[1]},
|
|
"mouse": {"x": pos[0], "y": pos[1]},
|
|
}
|
|
)
|
|
|
|
# Region helpers
|
|
elif action == "define_region":
|
|
if not name:
|
|
return json.dumps({"error": "name required"})
|
|
if x is None or y is None or width is None or height is None:
|
|
return json.dumps({"error": "x, y, width, height required"})
|
|
self._defined_regions[name] = (x, y, width, height)
|
|
return json.dumps(
|
|
{"success": True, "defined": name, "region": [x, y, width, height]}
|
|
)
|
|
|
|
elif action == "region_screenshot":
|
|
if not name:
|
|
return json.dumps({"error": "name required"})
|
|
if name not in self._defined_regions:
|
|
return json.dumps({"error": f"Region '{name}' not defined"})
|
|
reg = list(self._defined_regions[name])
|
|
data = await run(NativeControl.screenshot_native, reg)
|
|
return _result_to_mcp(
|
|
capture(
|
|
data,
|
|
fmt="png",
|
|
alt=f"region {name}",
|
|
max_width=max_width,
|
|
quality=quality,
|
|
full_res=full_res,
|
|
)
|
|
)
|
|
|
|
elif action == "region_locate":
|
|
if not name:
|
|
return json.dumps({"error": "name required"})
|
|
if not image_path and not text:
|
|
return json.dumps({"error": "image_path required"})
|
|
if name not in self._defined_regions:
|
|
return json.dumps({"error": f"Region '{name}' not defined"})
|
|
path = image_path or text
|
|
reg = self._defined_regions[name]
|
|
result = await run(self._locate, path, confidence, reg)
|
|
return result
|
|
|
|
# Timing
|
|
elif action == "sleep":
|
|
if value is None:
|
|
return json.dumps({"error": "value required"})
|
|
await asyncio.sleep(float(value))
|
|
return json.dumps({"success": True, "slept": value})
|
|
|
|
elif action == "countdown":
|
|
if value is None:
|
|
return json.dumps({"error": "value required"})
|
|
for i in range(int(value), 0, -1):
|
|
await asyncio.sleep(1)
|
|
return json.dumps({"success": True, "countdown": value})
|
|
|
|
elif action == "set_pause":
|
|
if value is None:
|
|
return json.dumps({"error": "value required"})
|
|
self._pause = float(value)
|
|
if self._pyautogui:
|
|
self._pyautogui.PAUSE = self._pause
|
|
return json.dumps({"success": True, "pause": self._pause})
|
|
|
|
elif action == "set_failsafe":
|
|
if value is None:
|
|
return json.dumps({"error": "value required"})
|
|
self._failsafe = bool(value)
|
|
if self._pyautogui:
|
|
self._pyautogui.FAILSAFE = self._failsafe
|
|
return json.dumps({"success": True, "failsafe": self._failsafe})
|
|
|
|
# Batch operations
|
|
elif action == "batch":
|
|
if not actions:
|
|
return json.dumps({"error": "actions required"})
|
|
|
|
results = []
|
|
start = time.time()
|
|
|
|
for i, act in enumerate(actions):
|
|
act_type = act.get("action", "")
|
|
try:
|
|
if act_type == "click":
|
|
NativeControl.click(
|
|
act.get("x", 0),
|
|
act.get("y", 0),
|
|
act.get("button", "left"),
|
|
)
|
|
elif act_type == "type":
|
|
NativeControl.type_text(
|
|
act.get("text", ""), act.get("interval", 0)
|
|
)
|
|
elif act_type == "press":
|
|
NativeControl.press(act.get("key", ""))
|
|
elif act_type == "hotkey":
|
|
NativeControl.hotkey(*act.get("keys", []))
|
|
elif act_type == "move":
|
|
NativeControl.move(act.get("x", 0), act.get("y", 0))
|
|
elif act_type == "scroll":
|
|
NativeControl.scroll(
|
|
act.get("amount", 0), act.get("x"), act.get("y")
|
|
)
|
|
elif act_type == "sleep":
|
|
time.sleep(act.get("ms", 0) / 1000)
|
|
else:
|
|
results.append(
|
|
{"index": i, "action": act_type, "error": "unknown"}
|
|
)
|
|
continue
|
|
results.append(
|
|
{"index": i, "action": act_type, "success": True}
|
|
)
|
|
except Exception as e:
|
|
results.append(
|
|
{"index": i, "action": act_type, "error": str(e)}
|
|
)
|
|
|
|
elapsed = time.time() - start
|
|
return json.dumps(
|
|
{
|
|
"success": True,
|
|
"count": len(results),
|
|
"elapsed_ms": round(elapsed * 1000, 2),
|
|
"results": results,
|
|
}
|
|
)
|
|
|
|
# Info
|
|
elif action == "info":
|
|
pos = NativeControl.mouse_position()
|
|
size = NativeControl.screen_size()
|
|
platform_info = NativeControl.get_platform_info()
|
|
return json.dumps(
|
|
{
|
|
"screen": {"width": size[0], "height": size[1]},
|
|
"mouse": {"x": pos[0], "y": pos[1]},
|
|
"platform": platform_info,
|
|
"pause": self._pause,
|
|
"failsafe": self._failsafe,
|
|
"regions": list(self._defined_regions.keys()),
|
|
}
|
|
)
|
|
|
|
elif action == "position":
|
|
pos = NativeControl.mouse_position()
|
|
return json.dumps({"x": pos[0], "y": pos[1]})
|
|
|
|
else:
|
|
return json.dumps({"error": f"Unknown action: {action}"})
|
|
|
|
except Exception as e:
|
|
return json.dumps({"error": str(e)})
|
|
|
|
@staticmethod
|
|
def _destination(name: str | None) -> str | None:
|
|
"""Resolve a caller-supplied screenshot name to an absolute .png path."""
|
|
if not name:
|
|
return None
|
|
p = os.path.expanduser(name)
|
|
if not os.path.isabs(p):
|
|
p = os.path.join(tempfile.gettempdir(), p)
|
|
return p if p.endswith(".png") else p + ".png"
|
|
|
|
# Helper methods for pyautogui-only features
|
|
|
|
def _locate(self, image_path: str, confidence: float, region: tuple | None) -> str:
|
|
pg = self._ensure_pyautogui()
|
|
path = Path(image_path)
|
|
if not path.exists():
|
|
return json.dumps({"error": f"Image not found: {image_path}"})
|
|
try:
|
|
kwargs: dict[str, Any] = {}
|
|
if confidence < 1.0:
|
|
kwargs["confidence"] = confidence
|
|
if region:
|
|
kwargs["region"] = region
|
|
location = pg.locateOnScreen(str(path), **kwargs)
|
|
if location:
|
|
center = pg.center(location)
|
|
return json.dumps(
|
|
{
|
|
"found": True,
|
|
"center": {"x": center.x, "y": center.y},
|
|
"box": {
|
|
"left": location.left,
|
|
"top": location.top,
|
|
"width": location.width,
|
|
"height": location.height,
|
|
},
|
|
}
|
|
)
|
|
return json.dumps({"found": False})
|
|
except Exception as e:
|
|
return json.dumps({"error": str(e)})
|
|
|
|
def _locate_all(self, image_path: str, confidence: float) -> str:
|
|
pg = self._ensure_pyautogui()
|
|
path = Path(image_path)
|
|
if not path.exists():
|
|
return json.dumps({"error": f"Image not found: {image_path}"})
|
|
try:
|
|
kwargs: dict[str, Any] = {}
|
|
if confidence < 1.0:
|
|
kwargs["confidence"] = confidence
|
|
locations = list(pg.locateAllOnScreen(str(path), **kwargs))
|
|
results = []
|
|
for loc in locations:
|
|
center = pg.center(loc)
|
|
results.append(
|
|
{
|
|
"center": {"x": center.x, "y": center.y},
|
|
"box": {
|
|
"left": loc.left,
|
|
"top": loc.top,
|
|
"width": loc.width,
|
|
"height": loc.height,
|
|
},
|
|
}
|
|
)
|
|
return json.dumps({"found": len(results), "locations": results})
|
|
except Exception as e:
|
|
return json.dumps({"error": str(e)})
|
|
|
|
def _locate_center(self, image_path: str, confidence: float) -> str:
|
|
pg = self._ensure_pyautogui()
|
|
path = Path(image_path)
|
|
if not path.exists():
|
|
return json.dumps({"error": f"Image not found: {image_path}"})
|
|
try:
|
|
kwargs: dict[str, Any] = {}
|
|
if confidence < 1.0:
|
|
kwargs["confidence"] = confidence
|
|
center = pg.locateCenterOnScreen(str(path), **kwargs)
|
|
if center:
|
|
return json.dumps({"found": True, "x": center.x, "y": center.y})
|
|
return json.dumps({"found": False})
|
|
except Exception as e:
|
|
return json.dumps({"error": str(e)})
|
|
|
|
async def _wait_for_image(
|
|
self, image_path: str, timeout: float, confidence: float, run
|
|
) -> str:
|
|
path = Path(image_path)
|
|
if not path.exists():
|
|
return json.dumps({"error": f"Image not found: {image_path}"})
|
|
start = time.time()
|
|
while time.time() - start < timeout:
|
|
result = await run(self._locate_center, str(path), confidence)
|
|
data = json.loads(result) if result.startswith("{") else {}
|
|
if data.get("found"):
|
|
return json.dumps(
|
|
{
|
|
"found": True,
|
|
"x": data["x"],
|
|
"y": data["y"],
|
|
"elapsed": round(time.time() - start, 2),
|
|
}
|
|
)
|
|
await asyncio.sleep(0.1)
|
|
return json.dumps({"found": False, "timeout": timeout})
|
|
|
|
async def _wait_while_image(
|
|
self, image_path: str, timeout: float, confidence: float, run
|
|
) -> str:
|
|
path = Path(image_path)
|
|
if not path.exists():
|
|
return json.dumps({"error": f"Image not found: {image_path}"})
|
|
start = time.time()
|
|
while time.time() - start < timeout:
|
|
result = await run(self._locate_center, str(path), confidence)
|
|
data = json.loads(result) if result.startswith("{") else {}
|
|
if not data.get("found"):
|
|
return json.dumps(
|
|
{"disappeared": True, "elapsed": round(time.time() - start, 2)}
|
|
)
|
|
await asyncio.sleep(0.1)
|
|
return json.dumps(
|
|
{"disappeared": False, "timeout": timeout, "still_visible": True}
|
|
)
|
|
|
|
def _get_pixel(self, x: int, y: int) -> str:
|
|
pg = self._ensure_pyautogui()
|
|
try:
|
|
screenshot = pg.screenshot(region=(x, y, 1, 1))
|
|
pixel = screenshot.getpixel((0, 0))
|
|
return json.dumps(
|
|
{"x": x, "y": y, "color": {"r": pixel[0], "g": pixel[1], "b": pixel[2]}}
|
|
)
|
|
except Exception as e:
|
|
return json.dumps({"error": str(e)})
|
|
|
|
def _pixel_matches(
|
|
self, x: int, y: int, color: tuple[int, int, int], tolerance: int
|
|
) -> str:
|
|
pg = self._ensure_pyautogui()
|
|
try:
|
|
screenshot = pg.screenshot(region=(x, y, 1, 1))
|
|
pixel = screenshot.getpixel((0, 0))
|
|
matches = all(abs(pixel[i] - color[i]) <= tolerance for i in range(3))
|
|
return json.dumps(
|
|
{
|
|
"matches": matches,
|
|
"expected": {"r": color[0], "g": color[1], "b": color[2]},
|
|
"actual": {"r": pixel[0], "g": pixel[1], "b": pixel[2]},
|
|
}
|
|
)
|
|
except Exception as e:
|
|
return json.dumps({"error": str(e)})
|
|
|
|
def _get_screens(self) -> str:
|
|
if IS_MACOS:
|
|
try:
|
|
result = subprocess.run(
|
|
["system_profiler", "SPDisplaysDataType", "-json"],
|
|
capture_output=True,
|
|
text=True,
|
|
timeout=5,
|
|
)
|
|
if result.returncode == 0:
|
|
data = json.loads(result.stdout)
|
|
displays = []
|
|
for gpu in data.get("SPDisplaysDataType", []):
|
|
for disp in gpu.get("spdisplays_ndrvs", []):
|
|
displays.append(
|
|
{
|
|
"name": disp.get("_name", "Unknown"),
|
|
"resolution": disp.get(
|
|
"_spdisplays_resolution", "Unknown"
|
|
),
|
|
"main": disp.get("spdisplays_main")
|
|
== "spdisplays_yes",
|
|
}
|
|
)
|
|
return json.dumps(displays)
|
|
except Exception:
|
|
pass
|
|
size = NativeControl.screen_size()
|
|
return json.dumps(
|
|
[{"name": "Primary", "resolution": f"{size[0]}x{size[1]}", "main": True}]
|
|
)
|
|
|
|
@override
|
|
def register(self, mcp_server: FastMCP) -> None:
|
|
"""Register this tool with the MCP server."""
|
|
tool_self = self
|
|
|
|
@mcp_server.tool(name=self.name, description=self.description)
|
|
async def computer(
|
|
action: Annotated[str, Field(description="Action to perform")] = "info",
|
|
x: Annotated[int | None, Field(description="X coordinate")] = None,
|
|
y: Annotated[int | None, Field(description="Y coordinate")] = None,
|
|
dx: Annotated[int | None, Field(description="Delta X")] = None,
|
|
dy: Annotated[int | None, Field(description="Delta Y")] = None,
|
|
end_x: Annotated[int | None, Field(description="End X for drag")] = None,
|
|
end_y: Annotated[int | None, Field(description="End Y for drag")] = None,
|
|
text: Annotated[str | None, Field(description="Text to type")] = None,
|
|
key: Annotated[str | None, Field(description="Key to press")] = None,
|
|
keys: Annotated[
|
|
list[str] | None, Field(description="Keys for hotkey")
|
|
] = None,
|
|
button: Annotated[str, Field(description="Mouse button")] = "left",
|
|
amount: Annotated[int | None, Field(description="Scroll amount")] = None,
|
|
duration: Annotated[float, Field(description="Duration")] = 0.25,
|
|
interval: Annotated[float, Field(description="Type interval")] = 0.02,
|
|
region: Annotated[
|
|
list[int] | None, Field(description="Region [x,y,w,h]")
|
|
] = None,
|
|
clear: Annotated[bool, Field(description="Clear before write")] = False,
|
|
image_path: Annotated[str | None, Field(description="Image path")] = None,
|
|
confidence: Annotated[float, Field(description="Match confidence")] = 0.9,
|
|
timeout: Annotated[float, Field(description="Wait timeout")] = 10.0,
|
|
color: Annotated[list[int] | None, Field(description="RGB color")] = None,
|
|
tolerance: Annotated[int, Field(description="Color tolerance")] = 0,
|
|
title: Annotated[str | None, Field(description="Window title")] = None,
|
|
use_regex: Annotated[bool, Field(description="Regex match")] = False,
|
|
name: Annotated[str | None, Field(description="Region name")] = None,
|
|
width: Annotated[int | None, Field(description="Width")] = None,
|
|
height: Annotated[int | None, Field(description="Height")] = None,
|
|
max_width: Annotated[
|
|
int,
|
|
Field(
|
|
description=(
|
|
"Longest edge of the inline screenshot, in pixels. 0 "
|
|
"inlines the capture at native resolution."
|
|
)
|
|
),
|
|
] = 1280,
|
|
quality: Annotated[
|
|
int, Field(description="Inline JPEG quality, 1-95.")
|
|
] = 70,
|
|
full_res: Annotated[
|
|
bool,
|
|
Field(
|
|
description=(
|
|
"Inline the native-resolution PNG. Costs 15-25x the "
|
|
"context; ask for it only when the pixels are the question."
|
|
)
|
|
),
|
|
] = False,
|
|
value: Annotated[float | None, Field(description="Value")] = None,
|
|
actions: Annotated[
|
|
list[dict] | None, Field(description="Batch actions")
|
|
] = None,
|
|
ctx: MCPContext = None,
|
|
) -> Any:
|
|
return await tool_self.call(
|
|
ctx,
|
|
action=action,
|
|
x=x,
|
|
y=y,
|
|
dx=dx,
|
|
dy=dy,
|
|
end_x=end_x,
|
|
end_y=end_y,
|
|
text=text,
|
|
key=key,
|
|
keys=keys,
|
|
button=button,
|
|
amount=amount,
|
|
duration=duration,
|
|
interval=interval,
|
|
region=region,
|
|
clear=clear,
|
|
image_path=image_path,
|
|
confidence=confidence,
|
|
timeout=timeout,
|
|
color=tuple(color) if color else None,
|
|
tolerance=tolerance,
|
|
title=title,
|
|
use_regex=use_regex,
|
|
name=name,
|
|
width=width,
|
|
height=height,
|
|
max_width=max_width,
|
|
quality=quality,
|
|
full_res=full_res,
|
|
value=value,
|
|
actions=actions,
|
|
)
|