mirror of
https://github.com/hanzoai/python-sdk.git
synced 2026-10-23 20:23:00 +00:00
zeekay
287c58a495
When name parameter is provided, screenshot saves to file and returns path
instead of base64 data. This prevents huge MCP responses.
Usage: computer(action='screenshot', name='screen.png')
Returns: {path: '/tmp/screen.png', size: ...}
1345 lines
55 KiB
Python
1345 lines
55 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 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,
|
|
)
|