"""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 io import os import re import sys import json import time import base64 import asyncio import shutil 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 # 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 Quartz import AppKit from Quartz import ( CGEventCreateMouseEvent, CGEventCreateKeyboardEvent, CGEventCreateScrollWheelEvent, CGEventPost, CGDisplayPixelsWide, CGDisplayPixelsHigh, CGMainDisplayID, kCGEventLeftMouseDown, kCGEventLeftMouseUp, kCGEventRightMouseDown, kCGEventRightMouseUp, kCGEventOtherMouseDown, kCGEventOtherMouseUp, kCGEventMouseMoved, kCGEventLeftMouseDragged, kCGEventRightMouseDragged, kCGEventOtherMouseDragged, kCGMouseButtonLeft, kCGMouseButtonRight, kCGMouseButtonCenter, kCGHIDEventTap, kCGScrollEventUnitLine, ) 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 # 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_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 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_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_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.""" 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.""" 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 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 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 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 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", # Keyboard "type", "write", "press", "key_down", "key_up", "hotkey", # Screen "screenshot", "screenshot_region", # 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", ] @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() 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, # Settings value: float | bool | None = None, # Batch actions: list[dict[str, Any]] | None = None, **kwargs, ) -> str: """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) # If name is provided, save to file instead of returning base64 if name: # Expand ~ and make absolute path file_path = os.path.expanduser(name) if not os.path.isabs(file_path): file_path = os.path.join(tempfile.gettempdir(), name) if not file_path.endswith('.png'): file_path += '.png' with open(file_path, 'wb') as f: f.write(data) return json.dumps({ "success": True, "format": "png", "size": len(data), "path": file_path, }) # Otherwise return base64 (large output warning) b64 = base64.b64encode(data).decode() return json.dumps({ "success": True, "format": "png", "size": len(data), "base64": b64, }) # 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) b64 = base64.b64encode(data).decode() return json.dumps({"success": True, "format": "png", "size": len(data), "base64": b64}) 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)}) # 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, value: Annotated[float | None, Field(description="Value")] = None, actions: Annotated[list[dict] | None, Field(description="Batch actions")] = None, ctx: MCPContext = None, ) -> str: 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, value=value, actions=actions, )