mirror of
https://github.com/hanzoai/python-sdk.git
synced 2026-10-23 20:23:00 +00:00
feat(computer): add ScreenTool for unified screen recording and Claude interpretation
New 'screen' tool provides one-shot recording → analysis → compression: screen(action="session", duration=30) # Record 30s, return compressed frames ACTIONS: - session: ONE-SHOT record → analyze → compress → return for Claude - capture: Single screenshot (optimized) - record: Start background recording - stop: Stop and process recording - analyze: Process existing video file HARD LIMITS (Claude API constraints): - Max 1568px per dimension (Claude 2000px limit for multi-image) - Max 100 images per request - Max 32MB payload DEFAULT SETTINGS: - 768px max dimension (good quality, safe margin) - 60% JPEG quality (aggressive compression) - 30 target frames per session - Activity detection @ 0.02 threshold USAGE: screen(action="session") # 30s → ~30 frames → ~500KB screen(action="capture") # Single screenshot screen(action="analyze", path="recording.mp4") # Process existing ENV VARS: HANZO_SCREEN_DURATION=30 HANZO_SCREEN_TARGET_FRAMES=30 HANZO_SCREEN_MAX_SIZE=768 HANZO_SCREEN_QUALITY=60 Bump hanzo-tools-computer to v0.5.0
This commit is contained in:
@@ -32,9 +32,10 @@ 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]
|
||||
TOOLS = [ComputerTool, FastComputerTool, VideoTool, MediaTool, ScreenTool]
|
||||
|
||||
__all__ = [
|
||||
"ComputerTool",
|
||||
@@ -45,6 +46,9 @@ __all__ = [
|
||||
"MediaResult",
|
||||
"ActivitySegment",
|
||||
"media_tool",
|
||||
"ScreenTool",
|
||||
"ScreenConfig",
|
||||
"screen_tool",
|
||||
"register_tools",
|
||||
"TOOLS",
|
||||
]
|
||||
|
||||
@@ -1,13 +1,18 @@
|
||||
"""Ultra-fast native macOS computer control using direct Quartz/CoreGraphics APIs.
|
||||
"""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 CGEvent posting (no pyautogui wrapper)
|
||||
- Direct native API calls (no pyautogui wrapper)
|
||||
- No PAUSE delays between operations
|
||||
- Native screencapture for screenshots (~50ms vs 200ms+ pyautogui)
|
||||
- Native screenshot tools (~50ms vs 200ms+ pyautogui)
|
||||
- Parallel batch operations
|
||||
- Event coalescing for rapid sequences
|
||||
|
||||
Performance characteristics:
|
||||
Performance characteristics (all platforms):
|
||||
- Click: <5ms
|
||||
- Type character: <2ms
|
||||
- Screenshot: <50ms (native) vs 200ms+ (pyautogui)
|
||||
@@ -23,12 +28,20 @@ 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
|
||||
|
||||
# Only import on macOS
|
||||
if sys.platform == "darwin":
|
||||
# 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
|
||||
@@ -61,8 +74,25 @@ if sys.platform == "darwin":
|
||||
QUARTZ_AVAILABLE = True
|
||||
except ImportError:
|
||||
QUARTZ_AVAILABLE = False
|
||||
else:
|
||||
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
|
||||
@@ -126,93 +156,252 @@ Action = Literal[
|
||||
|
||||
|
||||
class FastNativeControl:
|
||||
"""Zero-latency native macOS control using Quartz."""
|
||||
"""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."""
|
||||
loc = AppKit.NSEvent.mouseLocation()
|
||||
return int(loc.x), int(CGDisplayPixelsHigh(0) - loc.y)
|
||||
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."""
|
||||
return CGDisplayPixelsWide(CGMainDisplayID()), CGDisplayPixelsHigh(CGMainDisplayID())
|
||||
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(event_type: int, x: int, y: int, button: int = 0) -> None:
|
||||
"""Send mouse event directly via CGEvent - no delays."""
|
||||
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 button == "left":
|
||||
FastNativeControl._send_mouse_event(kCGEventLeftMouseDown, x, y, kCGMouseButtonLeft)
|
||||
FastNativeControl._send_mouse_event(kCGEventLeftMouseUp, x, y, kCGMouseButtonLeft)
|
||||
elif button == "right":
|
||||
FastNativeControl._send_mouse_event(kCGEventRightMouseDown, x, y, kCGMouseButtonRight)
|
||||
FastNativeControl._send_mouse_event(kCGEventRightMouseUp, x, y, kCGMouseButtonRight)
|
||||
elif button == "middle":
|
||||
FastNativeControl._send_mouse_event(kCGEventOtherMouseDown, x, y, kCGMouseButtonCenter)
|
||||
FastNativeControl._send_mouse_event(kCGEventOtherMouseUp, x, y, kCGMouseButtonCenter)
|
||||
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 - minimal delay."""
|
||||
FastNativeControl.click(x, y)
|
||||
time.sleep(0.01) # Minimal 10ms for double-click detection
|
||||
FastNativeControl.click(x, y)
|
||||
"""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."""
|
||||
FastNativeControl._send_mouse_event(kCGEventMouseMoved, x, y, 0)
|
||||
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."""
|
||||
# Mouse down at start
|
||||
if button == "left":
|
||||
FastNativeControl._send_mouse_event(kCGEventLeftMouseDown, start_x, start_y, kCGMouseButtonLeft)
|
||||
drag_type = kCGEventLeftMouseDragged
|
||||
up_type = kCGEventLeftMouseUp
|
||||
btn = kCGMouseButtonLeft
|
||||
elif button == "right":
|
||||
FastNativeControl._send_mouse_event(kCGEventRightMouseDown, start_x, start_y, kCGMouseButtonRight)
|
||||
drag_type = kCGEventRightMouseDragged
|
||||
up_type = kCGEventRightMouseUp
|
||||
btn = kCGMouseButtonRight
|
||||
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:
|
||||
FastNativeControl._send_mouse_event(kCGEventOtherMouseDown, start_x, start_y, kCGMouseButtonCenter)
|
||||
drag_type = kCGEventOtherMouseDragged
|
||||
up_type = kCGEventOtherMouseUp
|
||||
btn = kCGMouseButtonCenter
|
||||
|
||||
# Interpolate drag path
|
||||
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(drag_type, cx, cy, btn)
|
||||
time.sleep(0.001) # 1ms between drag events
|
||||
|
||||
# Mouse up at end
|
||||
FastNativeControl._send_mouse_event(up_type, end_x, end_y, btn)
|
||||
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 - zero delay."""
|
||||
"""Scroll."""
|
||||
if x is not None and y is not None:
|
||||
FastNativeControl.move(x, y)
|
||||
|
||||
event = CGEventCreateScrollWheelEvent(None, kCGScrollEventUnitLine, 1, amount)
|
||||
CGEventPost(kCGHIDEventTap, event)
|
||||
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(key_code: int, down: bool) -> None:
|
||||
"""Send key event - no delays."""
|
||||
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)
|
||||
|
||||
@@ -220,129 +409,264 @@ class FastNativeControl:
|
||||
def key_down(key: str) -> None:
|
||||
"""Press key down."""
|
||||
key_lower = key.lower()
|
||||
if key_lower in KEY_CODES:
|
||||
FastNativeControl._send_key_event(KEY_CODES[key_lower], True)
|
||||
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 key_lower in KEY_CODES:
|
||||
FastNativeControl._send_key_event(KEY_CODES[key_lower], False)
|
||||
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."""
|
||||
FastNativeControl.key_down(key)
|
||||
FastNativeControl.key_up(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."""
|
||||
# Press all keys down
|
||||
for key in keys:
|
||||
FastNativeControl.key_down(key)
|
||||
# Release in reverse
|
||||
for key in reversed(keys):
|
||||
FastNativeControl.key_up(key)
|
||||
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 char in SHIFT_CHARS:
|
||||
# Need shift
|
||||
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())
|
||||
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."""
|
||||
for char in text:
|
||||
FastNativeControl.type_char(char)
|
||||
if IS_LINUX and XDOTOOL_AVAILABLE:
|
||||
if interval > 0:
|
||||
time.sleep(interval)
|
||||
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 screencapture - fastest method."""
|
||||
"""Take screenshot using native tools - fastest method per platform."""
|
||||
with tempfile.NamedTemporaryFile(suffix=".png", delete=False) as f:
|
||||
tmp_path = f.name
|
||||
|
||||
try:
|
||||
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)
|
||||
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)
|
||||
|
||||
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
|
||||
|
||||
with open(tmp_path, "rb") as f:
|
||||
return f.read()
|
||||
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 name."""
|
||||
script = f'tell application "{title}" to activate'
|
||||
result = subprocess.run(
|
||||
["osascript", "-e", script],
|
||||
capture_output=True,
|
||||
timeout=5,
|
||||
)
|
||||
return result.returncode == 0
|
||||
"""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."""
|
||||
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]),
|
||||
}
|
||||
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 native macOS computer control.
|
||||
"""Ultra-fast cross-platform computer control.
|
||||
|
||||
Uses direct Quartz/CoreGraphics APIs for zero-latency operations.
|
||||
~10-50x faster than pyautogui for most operations.
|
||||
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"
|
||||
@@ -355,9 +679,13 @@ class FastComputerTool(BaseTool):
|
||||
@property
|
||||
@override
|
||||
def description(self) -> str:
|
||||
return """Ultra-fast native macOS computer control.
|
||||
platform_info = FastNativeControl.get_platform_info()
|
||||
return f"""Ultra-fast cross-platform computer control.
|
||||
|
||||
Uses direct Quartz/CoreGraphics APIs - 10-50x faster than standard pyautogui.
|
||||
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
|
||||
@@ -428,12 +756,7 @@ Examples:
|
||||
**kwargs,
|
||||
) -> str:
|
||||
"""Execute fast computer control action."""
|
||||
if sys.platform != "darwin":
|
||||
return json.dumps({"error": f"Only macOS supported, got {sys.platform}"})
|
||||
|
||||
if not QUARTZ_AVAILABLE:
|
||||
return json.dumps({"error": "Quartz not available. Install pyobjc-framework-Quartz"})
|
||||
|
||||
# Cross-platform support - we have backends for all platforms
|
||||
loop = asyncio.get_event_loop()
|
||||
|
||||
def run(fn, *args):
|
||||
@@ -540,9 +863,11 @@ Examples:
|
||||
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":
|
||||
|
||||
@@ -80,21 +80,46 @@ class MediaLimits:
|
||||
session_target_frames: int = 30 # Target frames for session summary
|
||||
session_compression_quality: int = 60 # Aggressive compression for sessions
|
||||
|
||||
# Hard limits (Claude API constraints for multi-image requests)
|
||||
# IMPORTANT: Claude has 2000px max for many-image requests
|
||||
# We use 1568px to stay well under the limit
|
||||
HARD_MAX_RESOLUTION: int = 1568 # Never exceed this
|
||||
HARD_MAX_IMAGES: int = 100
|
||||
HARD_MAX_PAYLOAD_MB: float = 32.0
|
||||
|
||||
@classmethod
|
||||
def from_env(cls) -> "MediaLimits":
|
||||
"""Load limits from environment variables."""
|
||||
"""Load limits from environment variables with hard cap enforcement."""
|
||||
return cls(
|
||||
max_images=int(os.environ.get("HANZO_MEDIA_MAX_IMAGES", "100")),
|
||||
max_payload_mb=float(os.environ.get("HANZO_MEDIA_MAX_PAYLOAD_MB", "32")),
|
||||
max_resolution=int(os.environ.get("HANZO_MEDIA_MAX_RESOLUTION", "1568")),
|
||||
optimal_size=int(os.environ.get("HANZO_MEDIA_OPTIMAL_SIZE", "768")),
|
||||
max_images=min(
|
||||
int(os.environ.get("HANZO_MEDIA_MAX_IMAGES", "100")),
|
||||
cls.HARD_MAX_IMAGES,
|
||||
),
|
||||
max_payload_mb=min(
|
||||
float(os.environ.get("HANZO_MEDIA_MAX_PAYLOAD_MB", "32")),
|
||||
cls.HARD_MAX_PAYLOAD_MB,
|
||||
),
|
||||
max_resolution=min(
|
||||
int(os.environ.get("HANZO_MEDIA_MAX_RESOLUTION", "1568")),
|
||||
cls.HARD_MAX_RESOLUTION, # Claude 2000px limit for multi-image
|
||||
),
|
||||
optimal_size=min(
|
||||
int(os.environ.get("HANZO_MEDIA_OPTIMAL_SIZE", "768")),
|
||||
cls.HARD_MAX_RESOLUTION,
|
||||
),
|
||||
jpeg_quality=int(os.environ.get("HANZO_MEDIA_JPEG_QUALITY", "85")),
|
||||
max_frames=int(os.environ.get("HANZO_MEDIA_MAX_FRAMES", "100")),
|
||||
max_frames=min(
|
||||
int(os.environ.get("HANZO_MEDIA_MAX_FRAMES", "100")),
|
||||
cls.HARD_MAX_IMAGES,
|
||||
),
|
||||
activity_threshold=float(os.environ.get("HANZO_MEDIA_ACTIVITY_THRESHOLD", "0.02")),
|
||||
min_activity_gap_ms=int(os.environ.get("HANZO_MEDIA_MIN_ACTIVITY_GAP_MS", "200")),
|
||||
scene_change_threshold=float(os.environ.get("HANZO_MEDIA_SCENE_THRESHOLD", "0.3")),
|
||||
session_max_duration=int(os.environ.get("HANZO_MEDIA_SESSION_MAX_DURATION", "60")),
|
||||
session_target_frames=int(os.environ.get("HANZO_MEDIA_SESSION_TARGET_FRAMES", "30")),
|
||||
session_target_frames=min(
|
||||
int(os.environ.get("HANZO_MEDIA_SESSION_TARGET_FRAMES", "30")),
|
||||
cls.HARD_MAX_IMAGES,
|
||||
),
|
||||
session_compression_quality=int(os.environ.get("HANZO_MEDIA_SESSION_QUALITY", "60")),
|
||||
)
|
||||
|
||||
|
||||
@@ -0,0 +1,662 @@
|
||||
"""Screen recording and AI interpretation tool for Hanzo AI.
|
||||
|
||||
One unified tool for capturing, recording, and sending screen data to Claude:
|
||||
- capture: Single screenshot
|
||||
- record: Start background recording
|
||||
- stop: Stop recording and get compressed frames
|
||||
- session: ONE-SHOT record → analyze → compress → return for Claude
|
||||
|
||||
The 'session' action is the primary interface:
|
||||
1. Records screen for specified duration (default 30s)
|
||||
2. Analyzes for activity (movement, scene changes)
|
||||
3. Extracts only keyframes at activity points
|
||||
4. Compresses heavily for minimal payload
|
||||
5. Returns frames ready for Claude interpretation
|
||||
|
||||
Typical usage:
|
||||
screen(action="session", duration=30) # 30s session
|
||||
# Returns ~30 compressed frames, ~500KB total
|
||||
# Claude can interpret the computer use session
|
||||
"""
|
||||
|
||||
import os
|
||||
import io
|
||||
import sys
|
||||
import json
|
||||
import time
|
||||
import base64
|
||||
import asyncio
|
||||
import tempfile
|
||||
import subprocess
|
||||
from typing import Any, Literal, Optional, Annotated, final, override
|
||||
from pathlib import Path
|
||||
from dataclasses import dataclass, field
|
||||
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
|
||||
|
||||
from .media_tool import (
|
||||
MediaLimits,
|
||||
MediaResult,
|
||||
ActivitySegment,
|
||||
_analyze_video_activity,
|
||||
_extract_activity_frames,
|
||||
_compress_session,
|
||||
_resize_image,
|
||||
)
|
||||
|
||||
# Thread pool for blocking operations
|
||||
_EXECUTOR = ThreadPoolExecutor(max_workers=2, thread_name_prefix="screen_")
|
||||
|
||||
|
||||
@dataclass
|
||||
class ScreenConfig:
|
||||
"""Configuration for screen capture sessions."""
|
||||
|
||||
# Session defaults
|
||||
default_duration: int = 30 # seconds
|
||||
max_duration: int = 120 # seconds
|
||||
|
||||
# Recording settings
|
||||
fps: int = 30
|
||||
quality: str = "medium" # low, medium, high
|
||||
|
||||
# Compression for Claude
|
||||
# IMPORTANT: Claude has 2000px max for multi-image requests
|
||||
# We use 768px for good quality while staying well under limit
|
||||
target_frames: int = 30
|
||||
max_size: int = 768 # pixels (hard capped at 1568 for safety)
|
||||
jpeg_quality: int = 60
|
||||
|
||||
# Activity detection
|
||||
activity_threshold: float = 0.02
|
||||
scene_threshold: float = 0.3
|
||||
|
||||
# Hard limits (Claude API constraints)
|
||||
HARD_MAX_SIZE: int = 1568 # Max dimension for multi-image (well under 2000)
|
||||
HARD_MAX_FRAMES: int = 100 # Max frames per request
|
||||
HARD_MAX_PAYLOAD_MB: float = 32.0 # Max total payload
|
||||
|
||||
@classmethod
|
||||
def from_env(cls) -> "ScreenConfig":
|
||||
"""Load from environment variables."""
|
||||
max_size = int(os.environ.get("HANZO_SCREEN_MAX_SIZE", "768"))
|
||||
# Enforce hard cap
|
||||
max_size = min(max_size, cls.HARD_MAX_SIZE)
|
||||
|
||||
return cls(
|
||||
default_duration=int(os.environ.get("HANZO_SCREEN_DURATION", "30")),
|
||||
max_duration=int(os.environ.get("HANZO_SCREEN_MAX_DURATION", "120")),
|
||||
target_frames=min(
|
||||
int(os.environ.get("HANZO_SCREEN_TARGET_FRAMES", "30")),
|
||||
cls.HARD_MAX_FRAMES,
|
||||
),
|
||||
max_size=max_size,
|
||||
jpeg_quality=int(os.environ.get("HANZO_SCREEN_QUALITY", "60")),
|
||||
activity_threshold=float(os.environ.get("HANZO_SCREEN_ACTIVITY_THRESHOLD", "0.02")),
|
||||
)
|
||||
|
||||
|
||||
def _capture_screenshot_native(region: Optional[list[int]] = None) -> bytes:
|
||||
"""Capture screenshot using macOS screencapture."""
|
||||
with tempfile.NamedTemporaryFile(suffix=".png", delete=False) as f:
|
||||
tmp_path = f.name
|
||||
|
||||
try:
|
||||
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)
|
||||
|
||||
with open(tmp_path, "rb") as f:
|
||||
return f.read()
|
||||
finally:
|
||||
if os.path.exists(tmp_path):
|
||||
os.unlink(tmp_path)
|
||||
|
||||
|
||||
def _record_screen(
|
||||
output_path: str,
|
||||
duration: int,
|
||||
fps: int = 30,
|
||||
quality: str = "medium",
|
||||
region: Optional[list[int]] = None,
|
||||
) -> str:
|
||||
"""Record screen to file for specified duration.
|
||||
|
||||
Returns path to recorded video file.
|
||||
"""
|
||||
quality_map = {"low": "28", "medium": "23", "high": "18"}
|
||||
crf = quality_map.get(quality, "23")
|
||||
|
||||
cmd = [
|
||||
"ffmpeg",
|
||||
"-y",
|
||||
"-f", "avfoundation",
|
||||
"-framerate", str(fps),
|
||||
"-i", "1:none", # Screen capture
|
||||
"-t", str(duration),
|
||||
"-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_path)
|
||||
|
||||
# Run and wait for completion
|
||||
subprocess.run(cmd, capture_output=True, timeout=duration + 30)
|
||||
|
||||
return output_path
|
||||
|
||||
|
||||
def _process_recording_for_claude(
|
||||
video_path: str,
|
||||
target_frames: int = 30,
|
||||
max_size: int = 512,
|
||||
quality: int = 60,
|
||||
activity_threshold: float = 0.02,
|
||||
scene_threshold: float = 0.3,
|
||||
) -> tuple[list[dict], list[dict], dict]:
|
||||
"""Process recorded video for Claude interpretation.
|
||||
|
||||
Returns (frames_with_base64, activity_segments, metadata).
|
||||
"""
|
||||
frames, segments, metadata = _compress_session(
|
||||
video_path,
|
||||
max_duration=120,
|
||||
target_frames=target_frames,
|
||||
activity_threshold=activity_threshold,
|
||||
scene_threshold=scene_threshold,
|
||||
max_size=max_size,
|
||||
quality=quality,
|
||||
)
|
||||
|
||||
# Convert to serializable format
|
||||
frames_data = []
|
||||
for data, info in frames:
|
||||
frames_data.append({
|
||||
"timestamp_sec": info["timestamp_sec"],
|
||||
"timestamp_ms": info["timestamp_ms"],
|
||||
"size": len(data),
|
||||
"base64": base64.b64encode(data).decode(),
|
||||
})
|
||||
|
||||
segments_data = [
|
||||
{
|
||||
"start_ms": s.start_ms,
|
||||
"end_ms": s.end_ms,
|
||||
"activity_score": round(s.activity_score, 4),
|
||||
}
|
||||
for s in segments
|
||||
]
|
||||
|
||||
return frames_data, segments_data, metadata
|
||||
|
||||
|
||||
Action = Literal[
|
||||
"capture", # Single screenshot
|
||||
"record", # Start recording (background)
|
||||
"stop", # Stop recording
|
||||
"status", # Recording status
|
||||
"session", # ONE-SHOT: record → analyze → compress → return
|
||||
"analyze", # Analyze existing video file
|
||||
"info", # System info
|
||||
]
|
||||
|
||||
|
||||
@final
|
||||
class ScreenTool(BaseTool):
|
||||
"""Screen recording and AI interpretation tool.
|
||||
|
||||
Primary action: 'session' - records screen, analyzes activity,
|
||||
compresses, and returns frames ready for Claude interpretation.
|
||||
|
||||
Usage:
|
||||
screen(action="session", duration=30)
|
||||
# Returns compressed keyframes from 30-second recording
|
||||
# Perfect for Claude to interpret computer use sessions
|
||||
"""
|
||||
|
||||
name = "screen"
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
permission_manager: Optional[PermissionManager] = None,
|
||||
config: Optional[ScreenConfig] = None,
|
||||
):
|
||||
if permission_manager is None:
|
||||
permission_manager = PermissionManager()
|
||||
self.permission_manager = permission_manager
|
||||
self.config = config or ScreenConfig.from_env()
|
||||
|
||||
# Recording state
|
||||
self._recording = False
|
||||
self._process: Optional[subprocess.Popen] = None
|
||||
self._output_file: Optional[str] = None
|
||||
self._start_time: float = 0
|
||||
|
||||
# Capabilities
|
||||
self._has_ffmpeg: Optional[bool] = None
|
||||
self._has_screencapture: Optional[bool] = None
|
||||
|
||||
def _check_ffmpeg(self) -> bool:
|
||||
if self._has_ffmpeg is None:
|
||||
try:
|
||||
subprocess.run(["ffmpeg", "-version"], capture_output=True, timeout=5)
|
||||
self._has_ffmpeg = True
|
||||
except Exception:
|
||||
self._has_ffmpeg = False
|
||||
return self._has_ffmpeg
|
||||
|
||||
def _check_screencapture(self) -> bool:
|
||||
if self._has_screencapture is None:
|
||||
self._has_screencapture = sys.platform == "darwin"
|
||||
return self._has_screencapture
|
||||
|
||||
@property
|
||||
@override
|
||||
def description(self) -> str:
|
||||
return f"""Screen recording and AI interpretation for Claude.
|
||||
|
||||
PRIMARY ACTION - session:
|
||||
Record screen → Analyze activity → Compress → Return for Claude
|
||||
|
||||
screen(action="session", duration=30)
|
||||
|
||||
This one call:
|
||||
1. Records screen for {self.config.default_duration}s (configurable)
|
||||
2. Detects activity (movement, clicks, typing)
|
||||
3. Extracts ~{self.config.target_frames} keyframes at activity points
|
||||
4. Compresses to ~{self.config.max_size}px @ {self.config.jpeg_quality}% quality
|
||||
5. Returns frames + activity analysis (~500KB total)
|
||||
|
||||
Perfect for Claude to interpret computer use sessions.
|
||||
|
||||
OTHER ACTIONS:
|
||||
|
||||
capture: Single screenshot
|
||||
- region: [x, y, width, height] (optional)
|
||||
- optimize: Compress for Claude (default: true)
|
||||
- Returns base64 image
|
||||
|
||||
record: Start background recording
|
||||
- duration: Recording duration in seconds (max {self.config.max_duration})
|
||||
- fps: Frame rate (default: 30)
|
||||
- quality: low/medium/high
|
||||
- region: Specific screen area
|
||||
|
||||
stop: Stop recording and process
|
||||
- Returns compressed frames like 'session'
|
||||
|
||||
status: Check recording state
|
||||
|
||||
analyze: Process existing video file
|
||||
- path: Video file path
|
||||
- Returns activity analysis + compressed frames
|
||||
|
||||
info: System capabilities
|
||||
|
||||
CONFIGURATION (env vars):
|
||||
HANZO_SCREEN_DURATION={self.config.default_duration} # Default session duration
|
||||
HANZO_SCREEN_TARGET_FRAMES={self.config.target_frames} # Target frames per session
|
||||
HANZO_SCREEN_MAX_SIZE={self.config.max_size} # Max frame dimension
|
||||
HANZO_SCREEN_QUALITY={self.config.jpeg_quality} # JPEG quality
|
||||
HANZO_SCREEN_ACTIVITY_THRESHOLD={self.config.activity_threshold}
|
||||
|
||||
EXAMPLES:
|
||||
screen(action="session") # 30-second session, returns ~30 compressed frames
|
||||
screen(action="session", duration=60) # 60-second session
|
||||
screen(action="capture") # Single screenshot
|
||||
screen(action="capture", region=[0, 0, 1920, 1080]) # Specific area
|
||||
screen(action="analyze", path="recording.mp4") # Process existing video
|
||||
"""
|
||||
|
||||
@override
|
||||
@auto_timeout("screen")
|
||||
async def call(
|
||||
self,
|
||||
ctx: MCPContext,
|
||||
action: str = "info",
|
||||
# Session/recording options
|
||||
duration: Optional[int] = None,
|
||||
fps: int = 30,
|
||||
quality: str = "medium",
|
||||
region: Optional[list[int]] = None,
|
||||
# Processing options
|
||||
target_frames: Optional[int] = None,
|
||||
max_size: Optional[int] = None,
|
||||
jpeg_quality: Optional[int] = None,
|
||||
activity_threshold: Optional[float] = None,
|
||||
# For analyze action
|
||||
path: Optional[str] = None,
|
||||
# Capture options
|
||||
optimize: bool = True,
|
||||
**kwargs,
|
||||
) -> str:
|
||||
"""Execute screen action."""
|
||||
if sys.platform != "darwin":
|
||||
return json.dumps({"error": "screen tool currently only supports macOS"})
|
||||
|
||||
loop = asyncio.get_event_loop()
|
||||
|
||||
# Apply config defaults with hard caps
|
||||
duration = duration or self.config.default_duration
|
||||
duration = min(duration, self.config.max_duration)
|
||||
target_frames = target_frames or self.config.target_frames
|
||||
target_frames = min(target_frames, ScreenConfig.HARD_MAX_FRAMES)
|
||||
max_size = max_size or self.config.max_size
|
||||
max_size = min(max_size, ScreenConfig.HARD_MAX_SIZE) # Claude 2000px limit
|
||||
jpeg_quality = jpeg_quality or self.config.jpeg_quality
|
||||
activity_threshold = activity_threshold or self.config.activity_threshold
|
||||
|
||||
try:
|
||||
if action == "info":
|
||||
return json.dumps({
|
||||
"platform": sys.platform,
|
||||
"has_ffmpeg": self._check_ffmpeg(),
|
||||
"has_screencapture": self._check_screencapture(),
|
||||
"recording": self._recording,
|
||||
"config": {
|
||||
"default_duration": self.config.default_duration,
|
||||
"max_duration": self.config.max_duration,
|
||||
"target_frames": self.config.target_frames,
|
||||
"max_size": self.config.max_size,
|
||||
"jpeg_quality": self.config.jpeg_quality,
|
||||
},
|
||||
})
|
||||
|
||||
elif action == "capture":
|
||||
# Single screenshot
|
||||
if not self._check_screencapture():
|
||||
return json.dumps({"error": "screencapture not available"})
|
||||
|
||||
data = await loop.run_in_executor(
|
||||
_EXECUTOR,
|
||||
_capture_screenshot_native,
|
||||
region,
|
||||
)
|
||||
|
||||
if optimize:
|
||||
# Compress for Claude
|
||||
data, info = await loop.run_in_executor(
|
||||
_EXECUTOR,
|
||||
_resize_image,
|
||||
data,
|
||||
max_size,
|
||||
jpeg_quality,
|
||||
True, # force JPEG
|
||||
)
|
||||
format_type = "jpeg"
|
||||
else:
|
||||
format_type = "png"
|
||||
info = {}
|
||||
|
||||
return json.dumps({
|
||||
"success": True,
|
||||
"format": format_type,
|
||||
"size": len(data),
|
||||
"dimensions": info.get("new_dimensions"),
|
||||
"base64": base64.b64encode(data).decode(),
|
||||
})
|
||||
|
||||
elif action == "session":
|
||||
# ONE-SHOT: record → analyze → compress → return
|
||||
if not self._check_ffmpeg():
|
||||
return json.dumps({"error": "FFmpeg required for screen recording"})
|
||||
|
||||
# Create temp file for recording
|
||||
with tempfile.NamedTemporaryFile(suffix=".mp4", delete=False) as f:
|
||||
video_path = f.name
|
||||
|
||||
try:
|
||||
# Record screen
|
||||
await loop.run_in_executor(
|
||||
_EXECUTOR,
|
||||
_record_screen,
|
||||
video_path,
|
||||
duration,
|
||||
fps,
|
||||
quality,
|
||||
region,
|
||||
)
|
||||
|
||||
# Process for Claude
|
||||
frames_data, segments_data, metadata = await loop.run_in_executor(
|
||||
_EXECUTOR,
|
||||
_process_recording_for_claude,
|
||||
video_path,
|
||||
target_frames,
|
||||
max_size,
|
||||
jpeg_quality,
|
||||
activity_threshold,
|
||||
self.config.scene_threshold,
|
||||
)
|
||||
|
||||
total_size = sum(f["size"] for f in frames_data)
|
||||
|
||||
return json.dumps({
|
||||
"success": True,
|
||||
"action": "session",
|
||||
"duration_seconds": duration,
|
||||
"frames": frames_data,
|
||||
"activity_segments": segments_data,
|
||||
"total_frames": len(frames_data),
|
||||
"total_size_bytes": total_size,
|
||||
"total_size_kb": round(total_size / 1024, 1),
|
||||
"compression_settings": {
|
||||
"target_frames": target_frames,
|
||||
"max_size": max_size,
|
||||
"quality": jpeg_quality,
|
||||
},
|
||||
"metadata": metadata,
|
||||
})
|
||||
|
||||
finally:
|
||||
# Cleanup temp file
|
||||
if os.path.exists(video_path):
|
||||
os.unlink(video_path)
|
||||
|
||||
elif action == "record":
|
||||
# Start background recording
|
||||
if self._recording:
|
||||
return json.dumps({"error": "Already recording"})
|
||||
|
||||
if not self._check_ffmpeg():
|
||||
return json.dumps({"error": "FFmpeg required"})
|
||||
|
||||
# Create output file
|
||||
self._output_file = tempfile.mktemp(suffix=".mp4")
|
||||
|
||||
# Build command
|
||||
quality_map = {"low": "28", "medium": "23", "high": "18"}
|
||||
crf = quality_map.get(quality, "23")
|
||||
|
||||
cmd = [
|
||||
"ffmpeg",
|
||||
"-y",
|
||||
"-f", "avfoundation",
|
||||
"-framerate", str(fps),
|
||||
"-i", "1:none",
|
||||
"-t", str(duration),
|
||||
"-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(self._output_file)
|
||||
|
||||
# Start recording process
|
||||
self._process = subprocess.Popen(
|
||||
cmd,
|
||||
stdin=subprocess.PIPE,
|
||||
stdout=subprocess.PIPE,
|
||||
stderr=subprocess.PIPE,
|
||||
)
|
||||
self._recording = True
|
||||
self._start_time = time.time()
|
||||
|
||||
return json.dumps({
|
||||
"success": True,
|
||||
"recording": True,
|
||||
"duration": duration,
|
||||
"output": self._output_file,
|
||||
})
|
||||
|
||||
elif action == "stop":
|
||||
# Stop recording and process
|
||||
if not self._recording:
|
||||
return json.dumps({"error": "Not recording"})
|
||||
|
||||
# Stop the process
|
||||
if self._process:
|
||||
try:
|
||||
self._process.stdin.write(b"q")
|
||||
self._process.stdin.flush()
|
||||
except Exception:
|
||||
pass
|
||||
await asyncio.sleep(0.5)
|
||||
self._process.terminate()
|
||||
try:
|
||||
self._process.wait(timeout=5)
|
||||
except subprocess.TimeoutExpired:
|
||||
self._process.kill()
|
||||
self._process = None
|
||||
|
||||
actual_duration = time.time() - self._start_time
|
||||
video_path = self._output_file
|
||||
|
||||
self._recording = False
|
||||
self._output_file = None
|
||||
self._start_time = 0
|
||||
|
||||
# Process the recording
|
||||
if video_path and os.path.exists(video_path):
|
||||
try:
|
||||
frames_data, segments_data, metadata = await loop.run_in_executor(
|
||||
_EXECUTOR,
|
||||
_process_recording_for_claude,
|
||||
video_path,
|
||||
target_frames,
|
||||
max_size,
|
||||
jpeg_quality,
|
||||
activity_threshold,
|
||||
self.config.scene_threshold,
|
||||
)
|
||||
|
||||
total_size = sum(f["size"] for f in frames_data)
|
||||
|
||||
return json.dumps({
|
||||
"success": True,
|
||||
"duration_seconds": round(actual_duration, 2),
|
||||
"frames": frames_data,
|
||||
"activity_segments": segments_data,
|
||||
"total_frames": len(frames_data),
|
||||
"total_size_kb": round(total_size / 1024, 1),
|
||||
"metadata": metadata,
|
||||
})
|
||||
|
||||
finally:
|
||||
os.unlink(video_path)
|
||||
else:
|
||||
return json.dumps({"error": "Recording file not found"})
|
||||
|
||||
elif action == "status":
|
||||
result = {
|
||||
"recording": self._recording,
|
||||
}
|
||||
if self._recording:
|
||||
result["duration_seconds"] = round(time.time() - self._start_time, 2)
|
||||
return json.dumps(result)
|
||||
|
||||
elif action == "analyze":
|
||||
# Analyze existing video file
|
||||
if not path:
|
||||
return json.dumps({"error": "path required"})
|
||||
|
||||
if not os.path.exists(path):
|
||||
return json.dumps({"error": f"File not found: {path}"})
|
||||
|
||||
frames_data, segments_data, metadata = await loop.run_in_executor(
|
||||
_EXECUTOR,
|
||||
_process_recording_for_claude,
|
||||
path,
|
||||
target_frames,
|
||||
max_size,
|
||||
jpeg_quality,
|
||||
activity_threshold,
|
||||
self.config.scene_threshold,
|
||||
)
|
||||
|
||||
total_size = sum(f["size"] for f in frames_data)
|
||||
|
||||
return json.dumps({
|
||||
"success": True,
|
||||
"source": path,
|
||||
"frames": frames_data,
|
||||
"activity_segments": segments_data,
|
||||
"total_frames": len(frames_data),
|
||||
"total_size_kb": round(total_size / 1024, 1),
|
||||
"metadata": metadata,
|
||||
})
|
||||
|
||||
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 screen(
|
||||
action: Annotated[str, Field(description="Action: session, capture, record, stop, analyze, info")] = "info",
|
||||
duration: Annotated[Optional[int], Field(description="Recording duration in seconds")] = None,
|
||||
fps: Annotated[int, Field(description="Frame rate for recording")] = 30,
|
||||
quality: Annotated[str, Field(description="Recording quality: low, medium, high")] = "medium",
|
||||
region: Annotated[Optional[list[int]], Field(description="Screen region [x, y, width, height]")] = None,
|
||||
target_frames: Annotated[Optional[int], Field(description="Target number of output frames")] = None,
|
||||
max_size: Annotated[Optional[int], Field(description="Max frame dimension in pixels")] = None,
|
||||
jpeg_quality: Annotated[Optional[int], Field(description="JPEG compression quality 1-100")] = None,
|
||||
activity_threshold: Annotated[Optional[float], Field(description="Activity detection sensitivity")] = None,
|
||||
path: Annotated[Optional[str], Field(description="Video file path for analyze action")] = None,
|
||||
optimize: Annotated[bool, Field(description="Optimize capture for Claude")] = True,
|
||||
ctx: MCPContext = None,
|
||||
) -> str:
|
||||
return await tool_self.call(
|
||||
ctx,
|
||||
action=action,
|
||||
duration=duration,
|
||||
fps=fps,
|
||||
quality=quality,
|
||||
region=region,
|
||||
target_frames=target_frames,
|
||||
max_size=max_size,
|
||||
jpeg_quality=jpeg_quality,
|
||||
activity_threshold=activity_threshold,
|
||||
path=path,
|
||||
optimize=optimize,
|
||||
)
|
||||
|
||||
|
||||
# Singleton instance
|
||||
screen_tool = ScreenTool()
|
||||
@@ -1,10 +1,9 @@
|
||||
"""Video capture and streaming tool for Hanzo AI.
|
||||
|
||||
High-performance screen recording with:
|
||||
- Native macOS AVFoundation capture (fastest)
|
||||
- FFmpeg fallback for cross-platform
|
||||
- Real-time frame streaming for AI analysis
|
||||
- Multiple output formats (H264, WebM, GIF)
|
||||
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
|
||||
@@ -12,11 +11,12 @@ Performance characteristics:
|
||||
- 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)
|
||||
- Max 10 frames per batch to fit context limits
|
||||
- JPEG compression (quality 85) for smaller payloads
|
||||
- Smart frame selection (detect changes, skip duplicates)
|
||||
- Estimated 100-200KB per frame after optimization
|
||||
- ~10-15MB typical for 100 frames with keyframe compression
|
||||
"""
|
||||
|
||||
import os
|
||||
@@ -243,31 +243,84 @@ Claude Vision Best Practices:
|
||||
"""
|
||||
|
||||
def _capture_frame_native(self, region: Optional[list[int]] = None) -> bytes:
|
||||
"""Capture frame using native macOS screencapture (fastest)."""
|
||||
"""Capture frame using native tools - cross-platform."""
|
||||
import shutil
|
||||
|
||||
with tempfile.NamedTemporaryFile(suffix=".png", delete=False) as f:
|
||||
tmp_path = f.name
|
||||
|
||||
try:
|
||||
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)
|
||||
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)
|
||||
|
||||
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
|
||||
|
||||
with open(tmp_path, "rb") as f:
|
||||
return f.read()
|
||||
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."""
|
||||
"""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",
|
||||
"-f", "avfoundation",
|
||||
"-i", "1:none", # Screen capture input
|
||||
*input_args,
|
||||
"-frames:v", "1",
|
||||
"-f", "image2pipe",
|
||||
"-vcodec", "png",
|
||||
@@ -276,8 +329,8 @@ Claude Vision Best Practices:
|
||||
|
||||
if region and len(region) == 4:
|
||||
x, y, w, h = region
|
||||
cmd.insert(-2, "-vf")
|
||||
cmd.insert(-2, f"crop={w}:{h}:{x}:{y}")
|
||||
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
|
||||
@@ -289,7 +342,7 @@ Claude Vision Best Practices:
|
||||
quality: str = "medium",
|
||||
region: Optional[list[int]] = None,
|
||||
) -> str:
|
||||
"""Start recording using FFmpeg."""
|
||||
"""Start recording using FFmpeg - cross-platform."""
|
||||
quality_map = {
|
||||
"low": "28",
|
||||
"medium": "23",
|
||||
@@ -297,12 +350,20 @@ Claude Vision Best Practices:
|
||||
}
|
||||
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
|
||||
"-f", "avfoundation",
|
||||
"-framerate", str(fps),
|
||||
"-i", "1:none",
|
||||
*input_args,
|
||||
"-c:v", "libx264",
|
||||
"-crf", crf,
|
||||
"-preset", "ultrafast",
|
||||
@@ -344,9 +405,7 @@ Claude Vision Best Practices:
|
||||
**kwargs,
|
||||
) -> str:
|
||||
"""Execute video action."""
|
||||
if sys.platform != "darwin":
|
||||
return f"Error: video tool currently only supports macOS"
|
||||
|
||||
# Cross-platform support via native tools + FFmpeg
|
||||
loop = asyncio.get_event_loop()
|
||||
|
||||
def run(fn, *args):
|
||||
|
||||
@@ -4,8 +4,8 @@ build-backend = "hatchling.build"
|
||||
|
||||
[project]
|
||||
name = "hanzo-tools-computer"
|
||||
version = "0.4.0"
|
||||
description = "Computer control tools for Hanzo AI - automation, video capture, media processing with activity detection (100 images/32MB limits)"
|
||||
version = "0.5.0"
|
||||
description = "Computer control tools for Hanzo AI - automation, screen recording, activity detection, session compression for Claude (100 images/32MB/1568px limits)"
|
||||
readme = "README.md"
|
||||
license = "MIT"
|
||||
requires-python = ">=3.12"
|
||||
|
||||
Generated
+1
-1
@@ -419,7 +419,7 @@ wheels = [
|
||||
|
||||
[[package]]
|
||||
name = "hanzo-tools-computer"
|
||||
version = "0.4.0"
|
||||
version = "0.5.0"
|
||||
source = { editable = "." }
|
||||
dependencies = [
|
||||
{ name = "hanzo-tools" },
|
||||
|
||||
Reference in New Issue
Block a user