mirror of
https://github.com/hanzoai/python-sdk.git
synced 2026-10-23 20:23:00 +00:00
fix(computer): correct docstring to show 2000px Claude limit
The docstring incorrectly showed 1568px as max dimension when the actual code uses 2000px (Claude's API limit).
This commit is contained in:
@@ -44,7 +44,7 @@ PACKAGE_TOOL_PREFIXES: dict[str, list[str]] = {
|
||||
"vector": ["index", "vector_index", "vector_search"],
|
||||
"config": ["config", "mode"],
|
||||
"mcp_tools": ["mcp"], # Consolidated MCP tool
|
||||
"computer": ["computer", "fast_computer", "video", "media", "screen"],
|
||||
"computer": ["computer", "screen"],
|
||||
}
|
||||
|
||||
|
||||
|
||||
@@ -1,10 +1,17 @@
|
||||
"""Computer control tools for Hanzo AI.
|
||||
|
||||
Tools:
|
||||
- computer: Control local computer via pyautogui (mouse, keyboard, screenshots)
|
||||
- fast_computer: Ultra-fast native macOS control via Quartz/CoreGraphics
|
||||
- video: Screen recording and video capture
|
||||
- media: Image/video processing with configurable limits (100 images, 32MB max)
|
||||
- computer: Unified computer control with native API acceleration
|
||||
- macOS: Quartz/CoreGraphics (10-50x faster than pyautogui)
|
||||
- Linux: xdotool/scrot for X11
|
||||
- Windows: win32api via ctypes
|
||||
- Fallback: pyautogui when native unavailable
|
||||
|
||||
- screen: Screen recording, capture, and Claude-optimized processing
|
||||
- capture: Single screenshot
|
||||
- record: Background recording
|
||||
- session: ONE-SHOT record → analyze → compress → return for Claude
|
||||
- Claude limits: 100 frames max, 32MB payload, 2000px max dimension
|
||||
|
||||
Install:
|
||||
pip install hanzo-tools-computer
|
||||
@@ -16,39 +23,35 @@ Usage:
|
||||
register_tools(mcp_server, permission_manager)
|
||||
|
||||
# Or access individual tools
|
||||
from hanzo_tools.computer import ComputerTool, FastComputerTool, VideoTool, MediaTool
|
||||
from hanzo_tools.computer import ComputerTool, ScreenTool
|
||||
|
||||
Media Limits (configurable via env vars):
|
||||
HANZO_MEDIA_MAX_IMAGES=100 # Max images per batch
|
||||
HANZO_MEDIA_MAX_PAYLOAD_MB=32 # Max total payload in MB
|
||||
HANZO_MEDIA_MAX_RESOLUTION=1568 # Max image dimension
|
||||
HANZO_MEDIA_OPTIMAL_SIZE=768 # Target size for optimization
|
||||
HANZO_MEDIA_JPEG_QUALITY=85 # JPEG quality (1-100)
|
||||
Screen Limits (configurable via env vars):
|
||||
HANZO_SCREEN_DURATION=30 # Default session duration (seconds)
|
||||
HANZO_SCREEN_TARGET_FRAMES=30 # Target frames per session
|
||||
HANZO_SCREEN_MAX_SIZE=768 # Max frame dimension (capped at 1568)
|
||||
HANZO_SCREEN_QUALITY=60 # JPEG quality (1-100)
|
||||
"""
|
||||
|
||||
from hanzo_tools.core import BaseTool, ToolRegistry, PermissionManager
|
||||
|
||||
from .computer_tool import ComputerTool
|
||||
from .fast_computer import FastComputerTool
|
||||
from .video_tool import VideoTool
|
||||
from .media_tool import MediaTool, MediaLimits, MediaResult, ActivitySegment, media_tool
|
||||
from .screen_tool import ScreenTool, ScreenConfig, screen_tool
|
||||
|
||||
# Export list for tool discovery
|
||||
TOOLS = [ComputerTool, FastComputerTool, VideoTool, MediaTool, ScreenTool]
|
||||
# Internal utilities (used by screen_tool)
|
||||
from .media_tool import MediaLimits, MediaResult, ActivitySegment, media_tool
|
||||
|
||||
# Export list for tool discovery - only 2 tools now
|
||||
TOOLS = [ComputerTool, ScreenTool]
|
||||
|
||||
__all__ = [
|
||||
"ComputerTool",
|
||||
"FastComputerTool",
|
||||
"VideoTool",
|
||||
"MediaTool",
|
||||
"ScreenTool",
|
||||
"ScreenConfig",
|
||||
"screen_tool",
|
||||
"MediaLimits",
|
||||
"MediaResult",
|
||||
"ActivitySegment",
|
||||
"media_tool",
|
||||
"ScreenTool",
|
||||
"ScreenConfig",
|
||||
"screen_tool",
|
||||
"register_tools",
|
||||
"TOOLS",
|
||||
]
|
||||
|
||||
File diff suppressed because it is too large
Load Diff
@@ -1,960 +0,0 @@
|
||||
"""Ultra-fast cross-platform computer control.
|
||||
|
||||
Platform-specific optimizations:
|
||||
- macOS: Direct Quartz/CoreGraphics APIs (fastest)
|
||||
- Linux: X11/xdotool or Wayland/ydotool
|
||||
- Windows: ctypes win32api (no pyautogui overhead)
|
||||
|
||||
Zero-latency design:
|
||||
- Direct native API calls (no pyautogui wrapper)
|
||||
- No PAUSE delays between operations
|
||||
- Native screenshot tools (~50ms vs 200ms+ pyautogui)
|
||||
- Parallel batch operations
|
||||
- Event coalescing for rapid sequences
|
||||
|
||||
Performance characteristics (all platforms):
|
||||
- Click: <5ms
|
||||
- Type character: <2ms
|
||||
- Screenshot: <50ms (native) vs 200ms+ (pyautogui)
|
||||
- Batch operations: Amortized to near-zero overhead
|
||||
"""
|
||||
|
||||
import io
|
||||
import os
|
||||
import sys
|
||||
import json
|
||||
import time
|
||||
import base64
|
||||
import asyncio
|
||||
import tempfile
|
||||
import subprocess
|
||||
import shutil
|
||||
from typing import Any, Literal, Optional, Annotated, final, override
|
||||
from pathlib import Path
|
||||
from concurrent.futures import ThreadPoolExecutor
|
||||
|
||||
# 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,
|
||||
CGEventSetIntegerValueField,
|
||||
kCGEventLeftMouseDown,
|
||||
kCGEventLeftMouseUp,
|
||||
kCGEventRightMouseDown,
|
||||
kCGEventRightMouseUp,
|
||||
kCGEventOtherMouseDown,
|
||||
kCGEventOtherMouseUp,
|
||||
kCGEventMouseMoved,
|
||||
kCGEventLeftMouseDragged,
|
||||
kCGEventRightMouseDragged,
|
||||
kCGEventOtherMouseDragged,
|
||||
kCGMouseButtonLeft,
|
||||
kCGMouseButtonRight,
|
||||
kCGMouseButtonCenter,
|
||||
kCGHIDEventTap,
|
||||
kCGScrollEventUnitLine,
|
||||
kCGKeyboardEventKeycode,
|
||||
)
|
||||
QUARTZ_AVAILABLE = True
|
||||
except ImportError:
|
||||
QUARTZ_AVAILABLE = False
|
||||
|
||||
# Linux: Check for X11 tools
|
||||
XDOTOOL_AVAILABLE = False
|
||||
SCROT_AVAILABLE = False
|
||||
GNOME_SCREENSHOT_AVAILABLE = False
|
||||
if IS_LINUX:
|
||||
XDOTOOL_AVAILABLE = shutil.which("xdotool") is not None
|
||||
SCROT_AVAILABLE = shutil.which("scrot") is not None
|
||||
GNOME_SCREENSHOT_AVAILABLE = shutil.which("gnome-screenshot") is not None
|
||||
|
||||
# Windows: ctypes for native API
|
||||
WIN32_AVAILABLE = False
|
||||
if IS_WINDOWS:
|
||||
try:
|
||||
import ctypes
|
||||
from ctypes import wintypes
|
||||
WIN32_AVAILABLE = True
|
||||
except ImportError:
|
||||
WIN32_AVAILABLE = False
|
||||
|
||||
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
|
||||
|
||||
# Shared thread pool
|
||||
_EXECUTOR = ThreadPoolExecutor(max_workers=4, thread_name_prefix="fast_computer_")
|
||||
|
||||
# Key mappings (same as pyautogui but with direct access)
|
||||
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, 'f13': 0x69, 'f14': 0x6b, 'f15': 0x71,
|
||||
'home': 0x73, 'end': 0x77, 'pageup': 0x74, 'pagedown': 0x79,
|
||||
'delete': 0x75, 'del': 0x75,
|
||||
'left': 0x7b, 'right': 0x7c, 'down': 0x7d, 'up': 0x7e,
|
||||
}
|
||||
|
||||
# Characters that need shift
|
||||
SHIFT_CHARS = '~!@#$%^&*()_+{}|:"<>?ABCDEFGHIJKLMNOPQRSTUVWXYZ'
|
||||
|
||||
|
||||
Action = Literal[
|
||||
# Mouse
|
||||
"click", "double_click", "right_click", "middle_click",
|
||||
"move", "drag", "scroll",
|
||||
# Keyboard
|
||||
"type", "press", "hotkey", "key_down", "key_up",
|
||||
# Screen
|
||||
"screenshot", "screenshot_region",
|
||||
# Info
|
||||
"info", "position", "size",
|
||||
# Batch
|
||||
"batch",
|
||||
# Window
|
||||
"focus_window", "get_active_window", "list_windows",
|
||||
]
|
||||
|
||||
|
||||
class FastNativeControl:
|
||||
"""Cross-platform zero-latency native control.
|
||||
|
||||
Platform backends:
|
||||
- macOS: Quartz/CoreGraphics (fastest)
|
||||
- Linux: xdotool/xte for X11
|
||||
- Windows: ctypes win32api
|
||||
"""
|
||||
|
||||
# ========== PLATFORM DETECTION ==========
|
||||
@staticmethod
|
||||
def get_platform_info() -> dict:
|
||||
"""Get current platform capabilities."""
|
||||
return {
|
||||
"platform": PLATFORM,
|
||||
"is_macos": IS_MACOS,
|
||||
"is_linux": IS_LINUX,
|
||||
"is_windows": IS_WINDOWS,
|
||||
"backends": {
|
||||
"quartz": QUARTZ_AVAILABLE,
|
||||
"xdotool": XDOTOOL_AVAILABLE,
|
||||
"scrot": SCROT_AVAILABLE,
|
||||
"win32": WIN32_AVAILABLE,
|
||||
}
|
||||
}
|
||||
|
||||
# ========== MOUSE POSITION ==========
|
||||
@staticmethod
|
||||
def mouse_position() -> tuple[int, int]:
|
||||
"""Get current 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
|
||||
)
|
||||
# Parse X=123\nY=456
|
||||
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:
|
||||
# Fallback to pyautogui
|
||||
try:
|
||||
import pyautogui
|
||||
return pyautogui.position()
|
||||
except:
|
||||
return (0, 0)
|
||||
|
||||
# ========== SCREEN SIZE ==========
|
||||
@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:
|
||||
pass
|
||||
return (1920, 1080) # Default fallback
|
||||
elif IS_WINDOWS and WIN32_AVAILABLE:
|
||||
return (
|
||||
ctypes.windll.user32.GetSystemMetrics(0),
|
||||
ctypes.windll.user32.GetSystemMetrics(1),
|
||||
)
|
||||
else:
|
||||
try:
|
||||
import pyautogui
|
||||
return pyautogui.size()
|
||||
except:
|
||||
return (1920, 1080)
|
||||
|
||||
# ========== MOUSE EVENTS (macOS) ==========
|
||||
@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)
|
||||
|
||||
# ========== CLICK ==========
|
||||
@staticmethod
|
||||
def click(x: int, y: int, button: str = "left") -> None:
|
||||
"""Click at position - zero delay."""
|
||||
if IS_MACOS and QUARTZ_AVAILABLE:
|
||||
if button == "left":
|
||||
FastNativeControl._send_mouse_event_macos(kCGEventLeftMouseDown, x, y, kCGMouseButtonLeft)
|
||||
FastNativeControl._send_mouse_event_macos(kCGEventLeftMouseUp, x, y, kCGMouseButtonLeft)
|
||||
elif button == "right":
|
||||
FastNativeControl._send_mouse_event_macos(kCGEventRightMouseDown, x, y, kCGMouseButtonRight)
|
||||
FastNativeControl._send_mouse_event_macos(kCGEventRightMouseUp, x, y, kCGMouseButtonRight)
|
||||
elif button == "middle":
|
||||
FastNativeControl._send_mouse_event_macos(kCGEventOtherMouseDown, x, y, kCGMouseButtonCenter)
|
||||
FastNativeControl._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:
|
||||
# Move mouse
|
||||
ctypes.windll.user32.SetCursorPos(x, y)
|
||||
# Click
|
||||
if button == "left":
|
||||
ctypes.windll.user32.mouse_event(0x0002, 0, 0, 0, 0) # MOUSEEVENTF_LEFTDOWN
|
||||
ctypes.windll.user32.mouse_event(0x0004, 0, 0, 0, 0) # MOUSEEVENTF_LEFTUP
|
||||
elif button == "right":
|
||||
ctypes.windll.user32.mouse_event(0x0008, 0, 0, 0, 0) # MOUSEEVENTF_RIGHTDOWN
|
||||
ctypes.windll.user32.mouse_event(0x0010, 0, 0, 0, 0) # MOUSEEVENTF_RIGHTUP
|
||||
elif button == "middle":
|
||||
ctypes.windll.user32.mouse_event(0x0020, 0, 0, 0, 0) # MOUSEEVENTF_MIDDLEDOWN
|
||||
ctypes.windll.user32.mouse_event(0x0040, 0, 0, 0, 0) # MOUSEEVENTF_MIDDLEUP
|
||||
else:
|
||||
try:
|
||||
import pyautogui
|
||||
pyautogui.click(x, y, button=button)
|
||||
except:
|
||||
pass
|
||||
|
||||
@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:
|
||||
FastNativeControl.click(x, y)
|
||||
time.sleep(0.01)
|
||||
FastNativeControl.click(x, y)
|
||||
|
||||
# ========== MOUSE MOVE ==========
|
||||
@staticmethod
|
||||
def move(x: int, y: int) -> None:
|
||||
"""Move mouse - zero delay."""
|
||||
if IS_MACOS and QUARTZ_AVAILABLE:
|
||||
FastNativeControl._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:
|
||||
try:
|
||||
import pyautogui
|
||||
pyautogui.moveTo(x, y, _pause=False)
|
||||
except:
|
||||
pass
|
||||
|
||||
# ========== DRAG ==========
|
||||
@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":
|
||||
FastNativeControl._send_mouse_event_macos(kCGEventLeftMouseDown, start_x, start_y, kCGMouseButtonLeft)
|
||||
drag_type = kCGEventLeftMouseDragged
|
||||
up_type = kCGEventLeftMouseUp
|
||||
btn = kCGMouseButtonLeft
|
||||
elif button == "right":
|
||||
FastNativeControl._send_mouse_event_macos(kCGEventRightMouseDown, start_x, start_y, kCGMouseButtonRight)
|
||||
drag_type = kCGEventRightMouseDragged
|
||||
up_type = kCGEventRightMouseUp
|
||||
btn = kCGMouseButtonRight
|
||||
else:
|
||||
FastNativeControl._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
|
||||
FastNativeControl._send_mouse_event_macos(drag_type, cx, cy, btn)
|
||||
time.sleep(0.001)
|
||||
FastNativeControl._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)
|
||||
if button == "left":
|
||||
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)
|
||||
if button == "left":
|
||||
ctypes.windll.user32.mouse_event(0x0004, 0, 0, 0, 0)
|
||||
else:
|
||||
try:
|
||||
import pyautogui
|
||||
pyautogui.moveTo(start_x, start_y, _pause=False)
|
||||
pyautogui.drag(end_x - start_x, end_y - start_y, _pause=False)
|
||||
except:
|
||||
pass
|
||||
|
||||
# ========== SCROLL ==========
|
||||
@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:
|
||||
FastNativeControl.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) # MOUSEEVENTF_WHEEL
|
||||
else:
|
||||
try:
|
||||
import pyautogui
|
||||
pyautogui.scroll(amount, _pause=False)
|
||||
except:
|
||||
pass
|
||||
|
||||
# ========== KEY EVENTS ==========
|
||||
@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:
|
||||
FastNativeControl._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 codes are different from macOS
|
||||
vk = _get_vk_code(key_lower)
|
||||
if vk:
|
||||
ctypes.windll.user32.keybd_event(vk, 0, 0, 0)
|
||||
else:
|
||||
try:
|
||||
import pyautogui
|
||||
pyautogui.keyDown(key_lower, _pause=False)
|
||||
except:
|
||||
pass
|
||||
|
||||
@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:
|
||||
FastNativeControl._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) # KEYEVENTF_KEYUP
|
||||
else:
|
||||
try:
|
||||
import pyautogui
|
||||
pyautogui.keyUp(key_lower, _pause=False)
|
||||
except:
|
||||
pass
|
||||
|
||||
@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:
|
||||
FastNativeControl.key_down(key)
|
||||
FastNativeControl.key_up(key)
|
||||
|
||||
@staticmethod
|
||||
def hotkey(*keys: str) -> None:
|
||||
"""Press key combination."""
|
||||
if IS_LINUX and XDOTOOL_AVAILABLE:
|
||||
# xdotool uses + for combo
|
||||
combo = "+".join(keys)
|
||||
subprocess.run(["xdotool", "key", combo], capture_output=True, timeout=2)
|
||||
else:
|
||||
for key in keys:
|
||||
FastNativeControl.key_down(key)
|
||||
for key in reversed(keys):
|
||||
FastNativeControl.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:
|
||||
FastNativeControl.key_down('shift')
|
||||
key = char.lower() if char.isalpha() else char
|
||||
if key in KEY_CODES:
|
||||
FastNativeControl.press(key)
|
||||
FastNativeControl.key_up('shift')
|
||||
elif char.lower() in KEY_CODES:
|
||||
FastNativeControl.press(char.lower())
|
||||
else:
|
||||
try:
|
||||
import pyautogui
|
||||
pyautogui.typewrite(char, _pause=False)
|
||||
except:
|
||||
pass
|
||||
|
||||
@staticmethod
|
||||
def type_text(text: str, interval: float = 0) -> None:
|
||||
"""Type text - fast mode with optional interval."""
|
||||
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:
|
||||
FastNativeControl.type_char(char)
|
||||
if interval > 0:
|
||||
time.sleep(interval)
|
||||
|
||||
@staticmethod
|
||||
def screenshot_native(region: list[int] | None = None) -> bytes:
|
||||
"""Take screenshot using native tools - fastest method per platform."""
|
||||
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:
|
||||
if 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 GNOME_SCREENSHOT_AVAILABLE:
|
||||
cmd = ["gnome-screenshot", "-f", tmp_path]
|
||||
if region and len(region) == 4:
|
||||
x, y, w, h = region
|
||||
cmd = ["gnome-screenshot", "-a", "-f", tmp_path]
|
||||
subprocess.run(cmd, capture_output=True, timeout=5)
|
||||
else:
|
||||
# Fallback: import using PIL
|
||||
try:
|
||||
from PIL import ImageGrab
|
||||
img = ImageGrab.grab(bbox=tuple(region) if region else None)
|
||||
img.save(tmp_path, "PNG")
|
||||
except:
|
||||
pass
|
||||
|
||||
elif IS_WINDOWS:
|
||||
# Use PIL ImageGrab on Windows (fastest)
|
||||
try:
|
||||
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")
|
||||
except:
|
||||
pass
|
||||
|
||||
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."""
|
||||
if IS_MACOS:
|
||||
script = f'tell application "{title}" to activate'
|
||||
result = subprocess.run(["osascript", "-e", script], capture_output=True, timeout=5)
|
||||
return result.returncode == 0
|
||||
elif IS_LINUX and XDOTOOL_AVAILABLE:
|
||||
result = subprocess.run(
|
||||
["xdotool", "search", "--name", title, "windowactivate"],
|
||||
capture_output=True, timeout=5
|
||||
)
|
||||
return result.returncode == 0
|
||||
elif IS_WINDOWS and WIN32_AVAILABLE:
|
||||
# Find and focus window by title
|
||||
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"}
|
||||
|
||||
|
||||
# Windows Virtual Key codes
|
||||
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())
|
||||
|
||||
|
||||
@final
|
||||
class FastComputerTool(BaseTool):
|
||||
"""Ultra-fast cross-platform computer control.
|
||||
|
||||
Platform backends:
|
||||
- macOS: Quartz/CoreGraphics APIs (~10-50x faster than pyautogui)
|
||||
- Linux: xdotool/scrot for X11 (fast native control)
|
||||
- Windows: win32api via ctypes (no pyautogui overhead)
|
||||
|
||||
Fallback: pyautogui if native APIs unavailable
|
||||
"""
|
||||
|
||||
name = "fast_computer"
|
||||
|
||||
def __init__(self, permission_manager: Optional[PermissionManager] = None):
|
||||
if permission_manager is None:
|
||||
permission_manager = PermissionManager()
|
||||
self.permission_manager = permission_manager
|
||||
|
||||
@property
|
||||
@override
|
||||
def description(self) -> str:
|
||||
platform_info = FastNativeControl.get_platform_info()
|
||||
return f"""Ultra-fast cross-platform computer control.
|
||||
|
||||
PLATFORM: {platform_info['platform']}
|
||||
BACKENDS: {', '.join(k for k, v in platform_info['backends'].items() if v) or 'pyautogui fallback'}
|
||||
|
||||
Uses native APIs for 10-50x faster operations than standard pyautogui.
|
||||
|
||||
MOUSE (< 5ms latency):
|
||||
- click(x, y, button): Click at position
|
||||
- double_click(x, y): Double click
|
||||
- right_click(x, y): Right click
|
||||
- middle_click(x, y): Middle click
|
||||
- move(x, y): Move cursor
|
||||
- drag(x, y, end_x, end_y): Drag operation
|
||||
- scroll(amount, x, y): Scroll wheel
|
||||
|
||||
KEYBOARD (< 2ms latency):
|
||||
- type(text): Type text (fast)
|
||||
- press(key): Press and release key
|
||||
- hotkey(keys): Key combination like ["command", "c"]
|
||||
- key_down(key) / key_up(key): Hold/release key
|
||||
|
||||
SCREEN (< 50ms):
|
||||
- screenshot(): Capture full screen
|
||||
- screenshot_region(region): Capture [x, y, w, h]
|
||||
|
||||
WINDOW:
|
||||
- focus_window(title): Activate window by app name
|
||||
- get_active_window(): Get active window info
|
||||
|
||||
INFO:
|
||||
- info(): Screen size and mouse position
|
||||
- position(): Mouse position
|
||||
- size(): Screen size
|
||||
|
||||
BATCH:
|
||||
- batch(actions): Execute multiple actions with minimal overhead
|
||||
|
||||
Examples:
|
||||
fast_computer(action="click", x=100, y=200)
|
||||
fast_computer(action="type", text="Hello World")
|
||||
fast_computer(action="hotkey", keys=["command", "c"])
|
||||
fast_computer(action="batch", actions=[
|
||||
{"action": "click", "x": 100, "y": 200},
|
||||
{"action": "type", "text": "test"},
|
||||
{"action": "press", "key": "return"}
|
||||
])
|
||||
"""
|
||||
|
||||
@override
|
||||
@auto_timeout("fast_computer")
|
||||
async def call(
|
||||
self,
|
||||
ctx: MCPContext,
|
||||
action: str = "info",
|
||||
# Coordinates
|
||||
x: int | None = None,
|
||||
y: 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,
|
||||
interval: float = 0,
|
||||
region: list[int] | None = None,
|
||||
# Window
|
||||
title: str | None = None,
|
||||
# Batch
|
||||
actions: list[dict[str, Any]] | None = None,
|
||||
**kwargs,
|
||||
) -> str:
|
||||
"""Execute fast computer control action."""
|
||||
# Cross-platform support - we have backends for all platforms
|
||||
loop = asyncio.get_event_loop()
|
||||
|
||||
def run(fn, *args):
|
||||
return loop.run_in_executor(_EXECUTOR, fn, *args)
|
||||
|
||||
try:
|
||||
# Mouse actions
|
||||
if action == "click":
|
||||
if x is None or y is None:
|
||||
return json.dumps({"error": "x and y required"})
|
||||
await run(FastNativeControl.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(FastNativeControl.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(FastNativeControl.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(FastNativeControl.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(FastNativeControl.move, x, y)
|
||||
return json.dumps({"success": True, "moved_to": [x, y]})
|
||||
|
||||
elif action == "drag":
|
||||
if x is None or y is None or end_x is None or end_y is None:
|
||||
return json.dumps({"error": "x, y, end_x, end_y required"})
|
||||
await run(FastNativeControl.drag, x, y, end_x, end_y, button)
|
||||
return json.dumps({"success": True, "dragged": {"from": [x, y], "to": [end_x, end_y]}})
|
||||
|
||||
elif action == "scroll":
|
||||
if amount is None:
|
||||
return json.dumps({"error": "amount required"})
|
||||
await run(FastNativeControl.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(FastNativeControl.type_text, text, interval)
|
||||
return json.dumps({"success": True, "typed": len(text)})
|
||||
|
||||
elif action == "press":
|
||||
if not key:
|
||||
return json.dumps({"error": "key required"})
|
||||
await run(FastNativeControl.press, key)
|
||||
return json.dumps({"success": True, "pressed": key})
|
||||
|
||||
elif action == "hotkey":
|
||||
if not keys:
|
||||
return json.dumps({"error": "keys required"})
|
||||
await run(FastNativeControl.hotkey, *keys)
|
||||
return json.dumps({"success": True, "hotkey": "+".join(keys)})
|
||||
|
||||
elif action == "key_down":
|
||||
if not key:
|
||||
return json.dumps({"error": "key required"})
|
||||
await run(FastNativeControl.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(FastNativeControl.key_up, key)
|
||||
return json.dumps({"success": True, "key_up": key})
|
||||
|
||||
# Screen actions
|
||||
elif action == "screenshot" or action == "screenshot_region":
|
||||
data = await run(FastNativeControl.screenshot_native, region)
|
||||
b64 = base64.b64encode(data).decode()
|
||||
return json.dumps({
|
||||
"success": True,
|
||||
"format": "png",
|
||||
"size": len(data),
|
||||
"base64": b64,
|
||||
})
|
||||
|
||||
# Window actions
|
||||
elif action == "focus_window":
|
||||
if not title:
|
||||
return json.dumps({"error": "title required"})
|
||||
success = await run(FastNativeControl.focus_window, title)
|
||||
return json.dumps({"success": success, "focused": title})
|
||||
|
||||
elif action == "get_active_window":
|
||||
result = await run(FastNativeControl.get_active_window)
|
||||
return json.dumps(result)
|
||||
|
||||
# Info actions
|
||||
elif action == "info":
|
||||
pos = FastNativeControl.mouse_position()
|
||||
size = FastNativeControl.screen_size()
|
||||
platform_info = FastNativeControl.get_platform_info()
|
||||
return json.dumps({
|
||||
"mouse": {"x": pos[0], "y": pos[1]},
|
||||
"screen": {"width": size[0], "height": size[1]},
|
||||
"platform": platform_info,
|
||||
})
|
||||
|
||||
elif action == "position":
|
||||
pos = FastNativeControl.mouse_position()
|
||||
return json.dumps({"x": pos[0], "y": pos[1]})
|
||||
|
||||
elif action == "size":
|
||||
size = FastNativeControl.screen_size()
|
||||
return json.dumps({"width": size[0], "height": size[1]})
|
||||
|
||||
# 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", "")
|
||||
|
||||
if act_type == "click":
|
||||
FastNativeControl.click(act.get("x", 0), act.get("y", 0), act.get("button", "left"))
|
||||
results.append({"index": i, "action": "click", "success": True})
|
||||
elif act_type == "type":
|
||||
FastNativeControl.type_text(act.get("text", ""), act.get("interval", 0))
|
||||
results.append({"index": i, "action": "type", "success": True})
|
||||
elif act_type == "press":
|
||||
FastNativeControl.press(act.get("key", ""))
|
||||
results.append({"index": i, "action": "press", "success": True})
|
||||
elif act_type == "hotkey":
|
||||
FastNativeControl.hotkey(*act.get("keys", []))
|
||||
results.append({"index": i, "action": "hotkey", "success": True})
|
||||
elif act_type == "move":
|
||||
FastNativeControl.move(act.get("x", 0), act.get("y", 0))
|
||||
results.append({"index": i, "action": "move", "success": True})
|
||||
elif act_type == "scroll":
|
||||
FastNativeControl.scroll(act.get("amount", 0), act.get("x"), act.get("y"))
|
||||
results.append({"index": i, "action": "scroll", "success": True})
|
||||
elif act_type == "sleep":
|
||||
time.sleep(act.get("ms", 0) / 1000)
|
||||
results.append({"index": i, "action": "sleep", "success": True})
|
||||
else:
|
||||
results.append({"index": i, "action": act_type, "error": "unknown action"})
|
||||
|
||||
elapsed = time.time() - start
|
||||
return json.dumps({
|
||||
"success": True,
|
||||
"count": len(results),
|
||||
"elapsed_ms": round(elapsed * 1000, 2),
|
||||
"results": results,
|
||||
})
|
||||
|
||||
else:
|
||||
return json.dumps({"error": f"Unknown action: {action}"})
|
||||
|
||||
except Exception as e:
|
||||
return json.dumps({"error": str(e)})
|
||||
|
||||
@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 fast_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,
|
||||
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,
|
||||
interval: Annotated[float, Field(description="Type interval")] = 0,
|
||||
region: Annotated[list[int] | None, Field(description="Screenshot region")] = None,
|
||||
title: Annotated[str | None, Field(description="Window title")] = 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, end_x=end_x, end_y=end_y,
|
||||
text=text, key=key, keys=keys,
|
||||
button=button, amount=amount, interval=interval,
|
||||
region=region, title=title, actions=actions,
|
||||
)
|
||||
@@ -1,785 +0,0 @@
|
||||
"""Video capture and streaming tool for Hanzo AI.
|
||||
|
||||
Cross-platform screen recording with:
|
||||
- macOS: Native screencapture + FFmpeg
|
||||
- Linux: scrot/gnome-screenshot + FFmpeg
|
||||
- Windows: PIL ImageGrab + FFmpeg
|
||||
|
||||
Performance characteristics:
|
||||
- Native capture: 60fps at 1080p with <10ms latency
|
||||
- Frame streaming: Individual frames for real-time AI vision
|
||||
- Background recording with async control
|
||||
|
||||
Claude Vision Optimization:
|
||||
- 100 frames max per request (Claude API limit)
|
||||
- 32MB max payload
|
||||
- Keyframe compression: every 5th frame at 85%, others at 60%
|
||||
- Frames resized to 768px (optimal for Claude vision)
|
||||
- Smart frame selection (detect changes, skip duplicates)
|
||||
- ~10-15MB typical for 100 frames with keyframe compression
|
||||
"""
|
||||
|
||||
import os
|
||||
import io
|
||||
import sys
|
||||
import json
|
||||
import time
|
||||
import base64
|
||||
import asyncio
|
||||
import tempfile
|
||||
import subprocess
|
||||
import hashlib
|
||||
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
|
||||
|
||||
# Thread pool for blocking operations
|
||||
_EXECUTOR = ThreadPoolExecutor(max_workers=2, thread_name_prefix="video_")
|
||||
|
||||
# Claude vision optimal settings (matches API limits)
|
||||
CLAUDE_MAX_FRAMES = 100 # Max frames per request (API limit)
|
||||
CLAUDE_MAX_PAYLOAD_MB = 32 # Max total payload in MB
|
||||
CLAUDE_OPTIMAL_SIZE = 768 # Optimal long edge size for speed/quality balance
|
||||
CLAUDE_JPEG_QUALITY_KEYFRAME = 85 # Quality for keyframes (every Nth frame)
|
||||
CLAUDE_JPEG_QUALITY_DELTA = 60 # Lower quality for in-between frames (Claude interpolates)
|
||||
CLAUDE_KEYFRAME_INTERVAL = 5 # Every 5th frame is a keyframe (higher quality)
|
||||
|
||||
Action = Literal[
|
||||
# Recording
|
||||
"start", # Start recording
|
||||
"stop", # Stop recording
|
||||
"status", # Get recording status
|
||||
"pause", # Pause recording
|
||||
"resume", # Resume recording
|
||||
# Capture
|
||||
"frame", # Capture single frame (fastest)
|
||||
"frames", # Capture N frames at interval
|
||||
"frames_claude", # Claude-optimized frame capture
|
||||
"stream", # Start frame streaming
|
||||
# Output
|
||||
"save", # Save to file
|
||||
"gif", # Convert to GIF
|
||||
# Info
|
||||
"info", # System info
|
||||
]
|
||||
|
||||
|
||||
def _resize_for_claude(
|
||||
image_data: bytes,
|
||||
max_size: int = CLAUDE_OPTIMAL_SIZE,
|
||||
quality: int = CLAUDE_JPEG_QUALITY_KEYFRAME,
|
||||
is_keyframe: bool = True,
|
||||
) -> bytes:
|
||||
"""Resize image for Claude vision API - optimal size and JPEG compression.
|
||||
|
||||
Args:
|
||||
image_data: Raw image bytes
|
||||
max_size: Max dimension (width or height)
|
||||
quality: JPEG quality (1-100)
|
||||
is_keyframe: If True, use higher quality. If False, use delta compression.
|
||||
|
||||
Keyframe strategy:
|
||||
- Every Nth frame is a keyframe (higher quality, ~100-200KB)
|
||||
- In-between frames use lower quality (~50-100KB) since Claude interpolates
|
||||
- This allows 100 frames at ~10MB total vs ~20MB with uniform quality
|
||||
"""
|
||||
try:
|
||||
from PIL import Image
|
||||
except ImportError:
|
||||
return image_data # Return original if PIL not available
|
||||
|
||||
# Use lower quality for non-keyframes (Claude can interpolate)
|
||||
actual_quality = quality if is_keyframe else min(quality, CLAUDE_JPEG_QUALITY_DELTA)
|
||||
|
||||
img = Image.open(io.BytesIO(image_data))
|
||||
|
||||
# Calculate new size maintaining aspect ratio
|
||||
width, height = img.size
|
||||
if max(width, height) > max_size:
|
||||
ratio = max_size / max(width, height)
|
||||
new_size = (int(width * ratio), int(height * ratio))
|
||||
img = img.resize(new_size, Image.Resampling.LANCZOS)
|
||||
|
||||
# Convert to JPEG for smaller size
|
||||
buffer = io.BytesIO()
|
||||
if img.mode in ('RGBA', 'P'):
|
||||
img = img.convert('RGB')
|
||||
img.save(buffer, format='JPEG', quality=actual_quality, optimize=True)
|
||||
|
||||
return buffer.getvalue()
|
||||
|
||||
|
||||
def _image_hash(image_data: bytes) -> str:
|
||||
"""Get perceptual hash of image for change detection."""
|
||||
return hashlib.md5(image_data[:1000]).hexdigest()[:8] # Quick hash of header
|
||||
|
||||
|
||||
def _frames_are_similar(frame1: bytes, frame2: bytes, threshold: float = 0.95) -> bool:
|
||||
"""Check if two frames are similar (for deduplication)."""
|
||||
# Simple size-based heuristic - if sizes are very similar, likely same
|
||||
if abs(len(frame1) - len(frame2)) / max(len(frame1), len(frame2)) < 0.05:
|
||||
return _image_hash(frame1) == _image_hash(frame2)
|
||||
return False
|
||||
|
||||
|
||||
@final
|
||||
class VideoTool(BaseTool):
|
||||
"""Screen recording and video capture.
|
||||
|
||||
Uses native macOS screencapture/AVFoundation for best performance.
|
||||
Falls back to FFmpeg for cross-platform support.
|
||||
"""
|
||||
|
||||
name = "video"
|
||||
|
||||
def __init__(self, permission_manager: Optional[PermissionManager] = None):
|
||||
if permission_manager is None:
|
||||
permission_manager = PermissionManager()
|
||||
self.permission_manager = permission_manager
|
||||
|
||||
# Recording state
|
||||
self._recording = False
|
||||
self._paused = False
|
||||
self._process: Optional[subprocess.Popen] = None
|
||||
self._output_file: Optional[str] = None
|
||||
self._start_time: float = 0
|
||||
self._frames: list[bytes] = []
|
||||
|
||||
# Streaming state
|
||||
self._streaming = False
|
||||
self._stream_task: Optional[asyncio.Task] = None
|
||||
self._frame_queue: asyncio.Queue[bytes] = asyncio.Queue(maxsize=30)
|
||||
|
||||
# Capabilities cache
|
||||
self._has_ffmpeg: Optional[bool] = None
|
||||
self._has_screencapture: Optional[bool] = None
|
||||
|
||||
def _check_ffmpeg(self) -> bool:
|
||||
"""Check if ffmpeg is available."""
|
||||
if self._has_ffmpeg is None:
|
||||
try:
|
||||
subprocess.run(["ffmpeg", "-version"], capture_output=True, timeout=5)
|
||||
self._has_ffmpeg = True
|
||||
except (subprocess.SubprocessError, FileNotFoundError):
|
||||
self._has_ffmpeg = False
|
||||
return self._has_ffmpeg
|
||||
|
||||
def _check_screencapture(self) -> bool:
|
||||
"""Check if macOS screencapture is available."""
|
||||
if self._has_screencapture is None:
|
||||
self._has_screencapture = sys.platform == "darwin"
|
||||
return self._has_screencapture
|
||||
|
||||
@property
|
||||
@override
|
||||
def description(self) -> str:
|
||||
return """Screen recording and video capture.
|
||||
|
||||
Actions:
|
||||
- start: Start screen recording
|
||||
- output: Output file path (optional, defaults to temp)
|
||||
- fps: Frames per second (default 30)
|
||||
- quality: Recording quality (low, medium, high)
|
||||
- region: [x, y, width, height] to record specific area
|
||||
|
||||
- stop: Stop recording and return file path
|
||||
|
||||
- status: Get current recording status
|
||||
|
||||
- pause/resume: Pause or resume recording
|
||||
|
||||
- frame: Capture single frame (fastest method)
|
||||
- Returns base64 PNG
|
||||
- Use for real-time AI vision
|
||||
|
||||
- frames: Capture multiple frames
|
||||
- count: Number of frames (default 10)
|
||||
- interval: Time between frames in ms (default 100)
|
||||
|
||||
- frames_claude: RECOMMENDED - Claude-optimized frame capture
|
||||
- Max 100 frames (Claude API limit)
|
||||
- Max 32MB total payload
|
||||
- Keyframe compression: every 5th frame at 85%, others at 60%
|
||||
- Claude interpolates between frames, so delta frames can be smaller
|
||||
- Automatically resizes to 768px (optimal for Claude vision)
|
||||
- Skips duplicate frames automatically
|
||||
- count: Number of frames (max 100)
|
||||
- interval: Time between frames in ms (default 100)
|
||||
|
||||
- stream: Start frame streaming
|
||||
- fps: Target frames per second (default 10)
|
||||
- Returns: stream_id for polling
|
||||
|
||||
- save: Save recording to specific path
|
||||
- output: Target file path
|
||||
- format: Output format (mp4, webm, mov)
|
||||
|
||||
- gif: Convert recording to GIF
|
||||
- output: Output path
|
||||
- fps: GIF frame rate (default 10)
|
||||
- width: Scale width (default 640)
|
||||
|
||||
Examples:
|
||||
video(action="start", fps=30, quality="high")
|
||||
video(action="frame") # Get single frame for AI analysis
|
||||
video(action="frames", count=5, interval=200) # 5 frames, 200ms apart
|
||||
video(action="frames_claude", count=100, interval=100) # 100 frames over 10s
|
||||
video(action="stop")
|
||||
video(action="gif", output="demo.gif")
|
||||
|
||||
Claude Vision Best Practices:
|
||||
- Use frames_claude for video analysis (auto-optimized)
|
||||
- 100 frames max, 32MB max payload
|
||||
- Keyframe strategy: every 5th frame is higher quality
|
||||
- Delta frames use 60% quality (Claude interpolates)
|
||||
- 100-500ms intervals capture meaningful changes
|
||||
- ~10-15MB typical for 100 frames with keyframe compression
|
||||
"""
|
||||
|
||||
def _capture_frame_native(self, region: Optional[list[int]] = None) -> bytes:
|
||||
"""Capture frame using native tools - cross-platform."""
|
||||
import shutil
|
||||
|
||||
with tempfile.NamedTemporaryFile(suffix=".png", delete=False) as f:
|
||||
tmp_path = f.name
|
||||
|
||||
try:
|
||||
if sys.platform == "darwin":
|
||||
# macOS: screencapture
|
||||
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=2)
|
||||
|
||||
elif sys.platform.startswith("linux"):
|
||||
# Linux: scrot or gnome-screenshot
|
||||
if shutil.which("scrot"):
|
||||
if region and len(region) == 4:
|
||||
x, y, w, h = region
|
||||
cmd = ["scrot", "-a", f"{x},{y},{w},{h}", "-o", tmp_path]
|
||||
else:
|
||||
cmd = ["scrot", "-o", tmp_path]
|
||||
subprocess.run(cmd, capture_output=True, timeout=2)
|
||||
elif shutil.which("gnome-screenshot"):
|
||||
cmd = ["gnome-screenshot", "-f", tmp_path]
|
||||
subprocess.run(cmd, capture_output=True, timeout=2)
|
||||
else:
|
||||
# Fallback to PIL
|
||||
try:
|
||||
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")
|
||||
except:
|
||||
pass
|
||||
|
||||
elif sys.platform == "win32":
|
||||
# Windows: PIL ImageGrab (fastest)
|
||||
try:
|
||||
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")
|
||||
except:
|
||||
pass
|
||||
|
||||
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)
|
||||
|
||||
def _capture_frame_ffmpeg(self, region: Optional[list[int]] = None) -> bytes:
|
||||
"""Capture frame using FFmpeg - cross-platform."""
|
||||
# Platform-specific input device
|
||||
if sys.platform == "darwin":
|
||||
input_args = ["-f", "avfoundation", "-i", "1:none"]
|
||||
elif sys.platform.startswith("linux"):
|
||||
# X11 capture
|
||||
input_args = ["-f", "x11grab", "-i", os.environ.get("DISPLAY", ":0")]
|
||||
elif sys.platform == "win32":
|
||||
input_args = ["-f", "gdigrab", "-i", "desktop"]
|
||||
else:
|
||||
return b""
|
||||
|
||||
cmd = [
|
||||
"ffmpeg",
|
||||
*input_args,
|
||||
"-frames:v", "1",
|
||||
"-f", "image2pipe",
|
||||
"-vcodec", "png",
|
||||
"-"
|
||||
]
|
||||
|
||||
if region and len(region) == 4:
|
||||
x, y, w, h = region
|
||||
cmd.insert(-4, "-vf")
|
||||
cmd.insert(-4, f"crop={w}:{h}:{x}:{y}")
|
||||
|
||||
result = subprocess.run(cmd, capture_output=True, timeout=5)
|
||||
return result.stdout
|
||||
|
||||
def _start_recording_ffmpeg(
|
||||
self,
|
||||
output: str,
|
||||
fps: int = 30,
|
||||
quality: str = "medium",
|
||||
region: Optional[list[int]] = None,
|
||||
) -> str:
|
||||
"""Start recording using FFmpeg - cross-platform."""
|
||||
quality_map = {
|
||||
"low": "28",
|
||||
"medium": "23",
|
||||
"high": "18",
|
||||
}
|
||||
crf = quality_map.get(quality, "23")
|
||||
|
||||
# Platform-specific input device
|
||||
if sys.platform == "darwin":
|
||||
input_args = ["-f", "avfoundation", "-framerate", str(fps), "-i", "1:none"]
|
||||
elif sys.platform.startswith("linux"):
|
||||
input_args = ["-f", "x11grab", "-framerate", str(fps), "-i", os.environ.get("DISPLAY", ":0")]
|
||||
elif sys.platform == "win32":
|
||||
input_args = ["-f", "gdigrab", "-framerate", str(fps), "-i", "desktop"]
|
||||
else:
|
||||
raise RuntimeError(f"Unsupported platform: {sys.platform}")
|
||||
|
||||
cmd = [
|
||||
"ffmpeg",
|
||||
"-y", # Overwrite
|
||||
*input_args,
|
||||
"-c:v", "libx264",
|
||||
"-crf", crf,
|
||||
"-preset", "ultrafast",
|
||||
"-pix_fmt", "yuv420p",
|
||||
]
|
||||
|
||||
if region and len(region) == 4:
|
||||
x, y, w, h = region
|
||||
cmd.extend(["-vf", f"crop={w}:{h}:{x}:{y}"])
|
||||
|
||||
cmd.append(output)
|
||||
|
||||
self._process = subprocess.Popen(
|
||||
cmd,
|
||||
stdin=subprocess.PIPE,
|
||||
stdout=subprocess.PIPE,
|
||||
stderr=subprocess.PIPE,
|
||||
)
|
||||
|
||||
return output
|
||||
|
||||
@override
|
||||
@auto_timeout("video")
|
||||
async def call(
|
||||
self,
|
||||
ctx: MCPContext,
|
||||
action: str = "info",
|
||||
# Recording options
|
||||
output: Optional[str] = None,
|
||||
fps: int = 30,
|
||||
quality: str = "medium",
|
||||
region: Optional[list[int]] = None,
|
||||
# Frame capture
|
||||
count: int = 10,
|
||||
interval: int = 100,
|
||||
# Output options
|
||||
format: str = "mp4",
|
||||
width: Optional[int] = None,
|
||||
**kwargs,
|
||||
) -> str:
|
||||
"""Execute video action."""
|
||||
# Cross-platform support via native tools + FFmpeg
|
||||
loop = asyncio.get_event_loop()
|
||||
|
||||
def run(fn, *args):
|
||||
return loop.run_in_executor(_EXECUTOR, fn, *args)
|
||||
|
||||
try:
|
||||
if action == "info":
|
||||
return json.dumps({
|
||||
"platform": sys.platform,
|
||||
"has_ffmpeg": self._check_ffmpeg(),
|
||||
"has_screencapture": self._check_screencapture(),
|
||||
"recording": self._recording,
|
||||
"paused": self._paused,
|
||||
"streaming": self._streaming,
|
||||
})
|
||||
|
||||
elif action == "frame":
|
||||
# Capture single frame - fastest method
|
||||
if self._check_screencapture():
|
||||
data = await run(self._capture_frame_native, region)
|
||||
elif self._check_ffmpeg():
|
||||
data = await run(self._capture_frame_ffmpeg, region)
|
||||
else:
|
||||
return "Error: No capture method available (need screencapture or ffmpeg)"
|
||||
|
||||
b64 = base64.b64encode(data).decode()
|
||||
return json.dumps({
|
||||
"success": True,
|
||||
"format": "png",
|
||||
"size": len(data),
|
||||
"base64": b64,
|
||||
})
|
||||
|
||||
elif action == "frames":
|
||||
# Capture multiple frames at interval
|
||||
frames_data = []
|
||||
for i in range(count):
|
||||
if self._check_screencapture():
|
||||
data = await run(self._capture_frame_native, region)
|
||||
else:
|
||||
data = await run(self._capture_frame_ffmpeg, region)
|
||||
|
||||
frames_data.append({
|
||||
"index": i,
|
||||
"size": len(data),
|
||||
"base64": base64.b64encode(data).decode(),
|
||||
})
|
||||
|
||||
if i < count - 1:
|
||||
await asyncio.sleep(interval / 1000)
|
||||
|
||||
return json.dumps({
|
||||
"success": True,
|
||||
"count": len(frames_data),
|
||||
"interval_ms": interval,
|
||||
"frames": frames_data,
|
||||
})
|
||||
|
||||
elif action == "frames_claude":
|
||||
# Claude-optimized frame capture with keyframe compression
|
||||
# - Max 100 frames (Claude API limit)
|
||||
# - Max 32MB total payload
|
||||
# - Keyframe strategy: every 5th frame at 85% quality, others at 60%
|
||||
# - Claude can interpolate between frames, so delta frames can be smaller
|
||||
# - Skip duplicate/similar frames
|
||||
actual_count = min(count, CLAUDE_MAX_FRAMES)
|
||||
max_payload_bytes = int(CLAUDE_MAX_PAYLOAD_MB * 1024 * 1024)
|
||||
frames_data = []
|
||||
last_frame_hash = None
|
||||
skipped = 0
|
||||
total_size = 0
|
||||
keyframes = 0
|
||||
delta_frames = 0
|
||||
|
||||
for i in range(actual_count * 2): # Allow for skipping duplicates
|
||||
if len(frames_data) >= actual_count:
|
||||
break
|
||||
|
||||
# Check payload limit (leave 1MB buffer)
|
||||
if total_size >= max_payload_bytes - 1024 * 1024:
|
||||
break
|
||||
|
||||
if self._check_screencapture():
|
||||
raw_data = await run(self._capture_frame_native, region)
|
||||
else:
|
||||
raw_data = await run(self._capture_frame_ffmpeg, region)
|
||||
|
||||
# Check for duplicate frames
|
||||
frame_hash = _image_hash(raw_data)
|
||||
if frame_hash == last_frame_hash:
|
||||
skipped += 1
|
||||
await asyncio.sleep(interval / 1000)
|
||||
continue
|
||||
|
||||
last_frame_hash = frame_hash
|
||||
|
||||
# Keyframe strategy: every Nth frame is higher quality
|
||||
frame_index = len(frames_data)
|
||||
is_keyframe = (frame_index % CLAUDE_KEYFRAME_INTERVAL == 0)
|
||||
|
||||
# Resize and compress for Claude with keyframe-aware quality
|
||||
optimized = await run(
|
||||
_resize_for_claude,
|
||||
raw_data,
|
||||
CLAUDE_OPTIMAL_SIZE,
|
||||
CLAUDE_JPEG_QUALITY_KEYFRAME,
|
||||
is_keyframe,
|
||||
)
|
||||
|
||||
# Check if this frame would exceed payload
|
||||
if total_size + len(optimized) > max_payload_bytes:
|
||||
break
|
||||
|
||||
if is_keyframe:
|
||||
keyframes += 1
|
||||
else:
|
||||
delta_frames += 1
|
||||
|
||||
frames_data.append({
|
||||
"index": frame_index,
|
||||
"timestamp_ms": i * interval,
|
||||
"is_keyframe": is_keyframe,
|
||||
"original_size": len(raw_data),
|
||||
"optimized_size": len(optimized),
|
||||
"format": "jpeg",
|
||||
"base64": base64.b64encode(optimized).decode(),
|
||||
})
|
||||
total_size += len(optimized)
|
||||
|
||||
if i < actual_count * 2 - 1:
|
||||
await asyncio.sleep(interval / 1000)
|
||||
|
||||
return json.dumps({
|
||||
"success": True,
|
||||
"count": len(frames_data),
|
||||
"max_frames": CLAUDE_MAX_FRAMES,
|
||||
"max_payload_mb": CLAUDE_MAX_PAYLOAD_MB,
|
||||
"optimal_size": CLAUDE_OPTIMAL_SIZE,
|
||||
"keyframes": keyframes,
|
||||
"delta_frames": delta_frames,
|
||||
"keyframe_quality": CLAUDE_JPEG_QUALITY_KEYFRAME,
|
||||
"delta_quality": CLAUDE_JPEG_QUALITY_DELTA,
|
||||
"skipped_duplicates": skipped,
|
||||
"total_size_kb": round(total_size / 1024, 1),
|
||||
"total_size_mb": round(total_size / (1024 * 1024), 2),
|
||||
"interval_ms": interval,
|
||||
"frames": frames_data,
|
||||
"note": f"100 frames max, 32MB limit. Keyframes (1/{CLAUDE_KEYFRAME_INTERVAL}) at {CLAUDE_JPEG_QUALITY_KEYFRAME}%, deltas at {CLAUDE_JPEG_QUALITY_DELTA}%",
|
||||
})
|
||||
|
||||
elif action == "start":
|
||||
if self._recording:
|
||||
return json.dumps({"error": "Already recording"})
|
||||
|
||||
if not self._check_ffmpeg():
|
||||
return json.dumps({"error": "FFmpeg required for video recording"})
|
||||
|
||||
# Generate output path
|
||||
if not output:
|
||||
output = tempfile.mktemp(suffix=f".{format}")
|
||||
|
||||
self._output_file = await run(
|
||||
self._start_recording_ffmpeg,
|
||||
output,
|
||||
fps,
|
||||
quality,
|
||||
region,
|
||||
)
|
||||
self._recording = True
|
||||
self._start_time = time.time()
|
||||
|
||||
return json.dumps({
|
||||
"success": True,
|
||||
"recording": True,
|
||||
"output": self._output_file,
|
||||
"fps": fps,
|
||||
"quality": quality,
|
||||
})
|
||||
|
||||
elif action == "stop":
|
||||
if not self._recording:
|
||||
return json.dumps({"error": "Not recording"})
|
||||
|
||||
if self._process:
|
||||
self._process.stdin.write(b"q")
|
||||
self._process.stdin.flush()
|
||||
await asyncio.sleep(0.5)
|
||||
self._process.terminate()
|
||||
try:
|
||||
self._process.wait(timeout=5)
|
||||
except subprocess.TimeoutExpired:
|
||||
self._process.kill()
|
||||
self._process = None
|
||||
|
||||
duration = time.time() - self._start_time
|
||||
output_file = self._output_file
|
||||
|
||||
self._recording = False
|
||||
self._paused = False
|
||||
self._output_file = None
|
||||
self._start_time = 0
|
||||
|
||||
# Get file size
|
||||
size = os.path.getsize(output_file) if output_file and os.path.exists(output_file) else 0
|
||||
|
||||
return json.dumps({
|
||||
"success": True,
|
||||
"output": output_file,
|
||||
"duration_seconds": round(duration, 2),
|
||||
"size_bytes": size,
|
||||
})
|
||||
|
||||
elif action == "status":
|
||||
result = {
|
||||
"recording": self._recording,
|
||||
"paused": self._paused,
|
||||
"streaming": self._streaming,
|
||||
}
|
||||
if self._recording:
|
||||
result["duration_seconds"] = round(time.time() - self._start_time, 2)
|
||||
result["output"] = self._output_file
|
||||
return json.dumps(result)
|
||||
|
||||
elif action == "pause":
|
||||
if not self._recording:
|
||||
return json.dumps({"error": "Not recording"})
|
||||
if self._paused:
|
||||
return json.dumps({"error": "Already paused"})
|
||||
|
||||
# Send SIGSTOP to pause
|
||||
if self._process:
|
||||
self._process.send_signal(19) # SIGSTOP
|
||||
self._paused = True
|
||||
|
||||
return json.dumps({"success": True, "paused": True})
|
||||
|
||||
elif action == "resume":
|
||||
if not self._recording:
|
||||
return json.dumps({"error": "Not recording"})
|
||||
if not self._paused:
|
||||
return json.dumps({"error": "Not paused"})
|
||||
|
||||
# Send SIGCONT to resume
|
||||
if self._process:
|
||||
self._process.send_signal(18) # SIGCONT
|
||||
self._paused = False
|
||||
|
||||
return json.dumps({"success": True, "resumed": True})
|
||||
|
||||
elif action == "gif":
|
||||
if not output:
|
||||
output = tempfile.mktemp(suffix=".gif")
|
||||
|
||||
if not self._output_file or not os.path.exists(self._output_file):
|
||||
return json.dumps({"error": "No recording to convert"})
|
||||
|
||||
gif_fps = fps or 10
|
||||
gif_width = width or 640
|
||||
|
||||
cmd = [
|
||||
"ffmpeg",
|
||||
"-y",
|
||||
"-i", self._output_file,
|
||||
"-vf", f"fps={gif_fps},scale={gif_width}:-1:flags=lanczos",
|
||||
"-loop", "0",
|
||||
output,
|
||||
]
|
||||
|
||||
result = subprocess.run(cmd, capture_output=True, timeout=60)
|
||||
|
||||
if result.returncode != 0:
|
||||
return json.dumps({"error": f"GIF conversion failed: {result.stderr.decode()}"})
|
||||
|
||||
return json.dumps({
|
||||
"success": True,
|
||||
"output": output,
|
||||
"size_bytes": os.path.getsize(output),
|
||||
})
|
||||
|
||||
elif action == "stream":
|
||||
if self._streaming:
|
||||
return json.dumps({"error": "Already streaming"})
|
||||
|
||||
self._streaming = True
|
||||
stream_fps = fps or 10
|
||||
|
||||
async def stream_frames():
|
||||
while self._streaming:
|
||||
try:
|
||||
if self._check_screencapture():
|
||||
data = await run(self._capture_frame_native, region)
|
||||
else:
|
||||
data = await run(self._capture_frame_ffmpeg, region)
|
||||
|
||||
try:
|
||||
self._frame_queue.put_nowait(data)
|
||||
except asyncio.QueueFull:
|
||||
# Drop oldest frame
|
||||
try:
|
||||
self._frame_queue.get_nowait()
|
||||
self._frame_queue.put_nowait(data)
|
||||
except asyncio.QueueEmpty:
|
||||
pass
|
||||
|
||||
await asyncio.sleep(1 / stream_fps)
|
||||
except Exception:
|
||||
break
|
||||
|
||||
self._stream_task = asyncio.create_task(stream_frames())
|
||||
|
||||
return json.dumps({
|
||||
"success": True,
|
||||
"streaming": True,
|
||||
"fps": stream_fps,
|
||||
"note": "Use frame action with stream_poll=true to get frames",
|
||||
})
|
||||
|
||||
elif action == "save":
|
||||
if not output:
|
||||
return json.dumps({"error": "output path required"})
|
||||
|
||||
if not self._output_file or not os.path.exists(self._output_file):
|
||||
return json.dumps({"error": "No recording to save"})
|
||||
|
||||
# Convert format if needed
|
||||
src_ext = Path(self._output_file).suffix
|
||||
dst_ext = Path(output).suffix
|
||||
|
||||
if src_ext == dst_ext:
|
||||
import shutil
|
||||
shutil.copy2(self._output_file, output)
|
||||
else:
|
||||
cmd = ["ffmpeg", "-y", "-i", self._output_file, output]
|
||||
subprocess.run(cmd, capture_output=True, timeout=60)
|
||||
|
||||
return json.dumps({
|
||||
"success": True,
|
||||
"output": output,
|
||||
"size_bytes": os.path.getsize(output),
|
||||
})
|
||||
|
||||
else:
|
||||
return json.dumps({"error": f"Unknown action: {action}"})
|
||||
|
||||
except Exception as e:
|
||||
return json.dumps({"error": str(e)})
|
||||
|
||||
@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 video(
|
||||
action: Annotated[str, Field(description="Action to perform")] = "info",
|
||||
output: Annotated[Optional[str], Field(description="Output file path")] = None,
|
||||
fps: Annotated[int, Field(description="Frames per second")] = 30,
|
||||
quality: Annotated[str, Field(description="Quality: low, medium, high")] = "medium",
|
||||
region: Annotated[Optional[list[int]], Field(description="Region [x,y,w,h]")] = None,
|
||||
count: Annotated[int, Field(description="Number of frames")] = 10,
|
||||
interval: Annotated[int, Field(description="Interval between frames (ms)")] = 100,
|
||||
format: Annotated[str, Field(description="Output format")] = "mp4",
|
||||
width: Annotated[Optional[int], Field(description="Scale width for GIF")] = None,
|
||||
ctx: MCPContext = None,
|
||||
) -> str:
|
||||
return await tool_self.call(
|
||||
ctx,
|
||||
action=action,
|
||||
output=output,
|
||||
fps=fps,
|
||||
quality=quality,
|
||||
region=region,
|
||||
count=count,
|
||||
interval=interval,
|
||||
format=format,
|
||||
width=width,
|
||||
)
|
||||
Reference in New Issue
Block a user