diff --git a/tools/capture/__init__.py b/tools/capture/__init__.py new file mode 100644 index 0000000..f153eba --- /dev/null +++ b/tools/capture/__init__.py @@ -0,0 +1 @@ +"""Screen capture tools — FFmpeg native recording and Cap integration.""" diff --git a/tools/capture/cap_recorder.py b/tools/capture/cap_recorder.py new file mode 100644 index 0000000..91168b8 --- /dev/null +++ b/tools/capture/cap_recorder.py @@ -0,0 +1,440 @@ +"""Cap integration tool — local Loom alternative. + +Detects whether Cap (https://cap.so) is installed, checks if it's running, +and picks up recordings from its output directory. If Cap isn't installed, +provides setup guidance the agent can present to the user. + +Cap provides: + - Polished recording UI with webcam overlay + - Cursor highlight and click effects + - Hardware-accelerated capture (GPU) + - Built-in editor with captions + - Clean system audio capture + +This tool does NOT control Cap directly — it acts as a bridge: + 1. Detect Cap installation and status + 2. Guide user through setup if needed + 3. Pick up completed recordings for the screen-demo pipeline +""" + +from __future__ import annotations + +import glob +import json +import os +import platform +import shutil +import subprocess +import time +from pathlib import Path +from typing import Any + +from tools.base_tool import ( + BaseTool, + Determinism, + ExecutionMode, + ResourceProfile, + ToolResult, + ToolRuntime, + ToolStability, + ToolTier, +) + + +def _find_cap_binary() -> str | None: + """Find the Cap executable on the system.""" + sys_platform = platform.system() + + if sys_platform == "Windows": + # Cap installs to AppData/Local on Windows + candidates = [ + Path(os.environ.get("LOCALAPPDATA", "")) / "Cap" / "Cap.exe", + Path(os.environ.get("PROGRAMFILES", "")) / "Cap" / "Cap.exe", + # Tauri apps sometimes install here + Path(os.environ.get("LOCALAPPDATA", "")) / "Programs" / "cap" / "Cap.exe", + ] + for c in candidates: + if c.exists(): + return str(c) + # Try PATH + if shutil.which("Cap") or shutil.which("cap"): + return shutil.which("Cap") or shutil.which("cap") + + elif sys_platform == "Darwin": + candidates = [ + Path("/Applications/Cap.app/Contents/MacOS/Cap"), + Path.home() / "Applications" / "Cap.app" / "Contents" / "MacOS" / "Cap", + ] + for c in candidates: + if c.exists(): + return str(c) + + elif sys_platform == "Linux": + candidates = ["cap", "Cap"] + for c in candidates: + found = shutil.which(c) + if found: + return found + # AppImage or Flatpak + appimage = Path.home() / "Applications" / "Cap.AppImage" + if appimage.exists(): + return str(appimage) + + return None + + +def _find_cap_recordings_dir() -> Path | None: + """Find Cap's recording output directory.""" + sys_platform = platform.system() + + if sys_platform == "Windows": + # Cap stores recordings in AppData + base = Path(os.environ.get("APPDATA", "")) / "so.cap.desktop" + if base.exists(): + return base + # Alternative path + base2 = Path(os.environ.get("LOCALAPPDATA", "")) / "so.cap.desktop" + if base2.exists(): + return base2 + + elif sys_platform == "Darwin": + base = Path.home() / "Library" / "Application Support" / "so.cap.desktop" + if base.exists(): + return base + + elif sys_platform == "Linux": + base = Path.home() / ".local" / "share" / "so.cap.desktop" + if base.exists(): + return base + + return None + + +def _is_cap_running() -> bool: + """Check if Cap is currently running.""" + sys_platform = platform.system() + try: + if sys_platform == "Windows": + result = subprocess.run( + ["tasklist", "/FI", "IMAGENAME eq Cap.exe", "/NH"], + capture_output=True, text=True, timeout=5, + ) + return "Cap.exe" in result.stdout + elif sys_platform == "Darwin": + result = subprocess.run( + ["pgrep", "-x", "Cap"], + capture_output=True, text=True, timeout=5, + ) + return result.returncode == 0 + elif sys_platform == "Linux": + result = subprocess.run( + ["pgrep", "-x", "cap"], + capture_output=True, text=True, timeout=5, + ) + return result.returncode == 0 + except Exception: + pass + return False + + +def _get_recent_recordings(recordings_dir: Path, since_seconds: int = 300) -> list[dict]: + """Find Cap recordings created within the last N seconds.""" + recordings = [] + cutoff = time.time() - since_seconds + + # Cap stores recordings as directories with output/ subdirs + for item in sorted(recordings_dir.iterdir(), key=lambda p: p.stat().st_mtime, reverse=True): + if not item.is_dir(): + continue + # Look for video files in the recording directory + for pattern in ["*.mp4", "output/*.mp4", "output/result.mp4"]: + for video in item.glob(pattern): + if video.stat().st_mtime > cutoff: + recordings.append({ + "path": str(video), + "name": item.name, + "size_mb": round(video.stat().st_size / (1024 * 1024), 1), + "modified": video.stat().st_mtime, + }) + + return recordings[:10] # Return most recent 10 + + +class CapRecorder(BaseTool): + name = "cap_recorder" + version = "0.1.0" + tier = ToolTier.SOURCE + capability = "screen_capture" + provider = "cap" + stability = ToolStability.BETA + execution_mode = ExecutionMode.SYNC + determinism = Determinism.DETERMINISTIC + runtime = ToolRuntime.LOCAL + + dependencies = [] # No hard dependencies — detection is graceful + install_instructions = ( + "Cap is a free, open-source Loom alternative.\n\n" + "Install from: https://cap.so/download\n" + " - Windows: Download and run the installer\n" + " - macOS: Download the .dmg or use: brew install --cask cap\n" + " - Linux: Download the AppImage from GitHub releases\n\n" + "Source code: https://github.com/CapSoftware/cap\n\n" + "Cap provides webcam overlay, cursor highlighting, and a polished\n" + "recording UI that FFmpeg-based recording cannot match." + ) + + capabilities = [ + "detect_cap", + "check_status", + "find_recordings", + "setup_guidance", + ] + + best_for = [ + "Professional screen recordings with webcam overlay", + "Cursor highlight and click effect recordings", + "Recording with a visual UI (not CLI-driven)", + "Recordings that need polished audio capture", + ] + + not_good_for = [ + "Automated/headless screen recording", + "Recording without user interaction", + "Quick recordings where setup time matters", + ] + + input_schema = { + "type": "object", + "required": ["operation"], + "properties": { + "operation": { + "type": "string", + "enum": ["detect", "status", "find_recordings", "setup_guide", "pick_latest"], + "description": ( + "'detect' — check if Cap is installed, " + "'status' — check if Cap is running, " + "'find_recordings' — list recent recordings, " + "'setup_guide' — get install instructions, " + "'pick_latest' — get the most recent recording file" + ), + }, + "output_dir": { + "type": "string", + "description": "For pick_latest: copy the recording here", + }, + "since_minutes": { + "type": "integer", + "default": 5, + "description": "For find_recordings: look back this many minutes", + }, + }, + } + + output_schema = { + "type": "object", + "properties": { + "installed": {"type": "boolean"}, + "running": {"type": "boolean"}, + "binary_path": {"type": ["string", "null"]}, + "recordings_dir": {"type": ["string", "null"]}, + "recordings": {"type": "array"}, + "setup_instructions": {"type": "string"}, + }, + } + + resource_profile = ResourceProfile( + cpu_cores=1, ram_mb=64, vram_mb=0, disk_mb=0, network_required=False, + ) + + side_effects = [] + fallback_tools = ["screen_recorder"] + + def get_status(self): + """Cap tool is always 'available' — it gracefully handles missing Cap.""" + from tools.base_tool import ToolStatus + return ToolStatus.AVAILABLE + + def execute(self, inputs: dict[str, Any]) -> ToolResult: + operation = inputs["operation"] + + if operation == "detect": + return self._detect() + elif operation == "status": + return self._status() + elif operation == "find_recordings": + since = inputs.get("since_minutes", 5) + return self._find_recordings(since) + elif operation == "setup_guide": + return self._setup_guide() + elif operation == "pick_latest": + output_dir = inputs.get("output_dir") + return self._pick_latest(output_dir) + else: + return ToolResult( + success=False, + error=f"Unknown operation: {operation}. " + f"Valid: detect, status, find_recordings, setup_guide, pick_latest", + ) + + def _detect(self) -> ToolResult: + binary = _find_cap_binary() + recordings_dir = _find_cap_recordings_dir() + running = _is_cap_running() if binary else False + + return ToolResult( + success=True, + data={ + "installed": binary is not None, + "running": running, + "binary_path": binary, + "recordings_dir": str(recordings_dir) if recordings_dir else None, + "platform": platform.system(), + }, + ) + + def _status(self) -> ToolResult: + binary = _find_cap_binary() + if not binary: + return ToolResult( + success=True, + data={ + "installed": False, + "running": False, + "message": "Cap is not installed. Use operation='setup_guide' for install instructions.", + }, + ) + + running = _is_cap_running() + recordings_dir = _find_cap_recordings_dir() + + return ToolResult( + success=True, + data={ + "installed": True, + "running": running, + "binary_path": binary, + "recordings_dir": str(recordings_dir) if recordings_dir else None, + "message": "Cap is running and ready to record." if running + else "Cap is installed but not running. The user should open Cap to start recording.", + }, + ) + + def _find_recordings(self, since_minutes: int) -> ToolResult: + recordings_dir = _find_cap_recordings_dir() + if not recordings_dir: + return ToolResult( + success=True, + data={ + "recordings": [], + "message": "Cap recordings directory not found. Cap may not be installed or hasn't made any recordings yet.", + }, + ) + + recordings = _get_recent_recordings(recordings_dir, since_seconds=since_minutes * 60) + + return ToolResult( + success=True, + data={ + "recordings": recordings, + "recordings_dir": str(recordings_dir), + "count": len(recordings), + "message": f"Found {len(recordings)} recording(s) from the last {since_minutes} minutes." + if recordings else f"No recordings found in the last {since_minutes} minutes.", + }, + ) + + def _setup_guide(self) -> ToolResult: + sys_platform = platform.system() + binary = _find_cap_binary() + + if binary: + return ToolResult( + success=True, + data={ + "installed": True, + "binary_path": binary, + "message": "Cap is already installed!", + "next_step": "Open Cap and start recording. When done, use operation='pick_latest' to grab the recording.", + }, + ) + + instructions = { + "Windows": { + "recommended": "Download from https://cap.so/download", + "alternative": "winget install CapSoftware.Cap", + "time_estimate": "2 minutes", + }, + "Darwin": { + "recommended": "brew install --cask cap", + "alternative": "Download .dmg from https://cap.so/download", + "time_estimate": "2 minutes", + }, + "Linux": { + "recommended": "Download AppImage from https://github.com/CapSoftware/cap/releases", + "alternative": "Build from source: https://github.com/CapSoftware/cap", + "time_estimate": "3-5 minutes", + }, + } + + platform_guide = instructions.get(sys_platform, instructions["Linux"]) + + return ToolResult( + success=True, + data={ + "installed": False, + "platform": sys_platform, + "setup": platform_guide, + "what_you_get": [ + "Webcam overlay (picture-in-picture)", + "Cursor highlight and click effects", + "Clean system + microphone audio capture", + "Built-in editor with auto-captions", + "Polished recording UI", + ], + "source_code": "https://github.com/CapSoftware/cap", + "message": f"Cap is not installed. Setup takes about {platform_guide['time_estimate']}.", + }, + ) + + def _pick_latest(self, output_dir: str | None) -> ToolResult: + recordings_dir = _find_cap_recordings_dir() + if not recordings_dir: + return ToolResult( + success=False, + error="Cap recordings directory not found.", + ) + + recordings = _get_recent_recordings(recordings_dir, since_seconds=3600) + if not recordings: + return ToolResult( + success=False, + error="No recent Cap recordings found. Record something in Cap first.", + ) + + latest = recordings[0] + source = Path(latest["path"]) + + if output_dir: + dest = Path(output_dir) / source.name + dest.parent.mkdir(parents=True, exist_ok=True) + shutil.copy2(source, dest) + return ToolResult( + success=True, + data={ + "output_path": str(dest), + "original_path": str(source), + "size_mb": latest["size_mb"], + "capture_method": "cap", + }, + artifacts=[str(dest)], + ) + + return ToolResult( + success=True, + data={ + "output_path": str(source), + "size_mb": latest["size_mb"], + "capture_method": "cap", + }, + artifacts=[str(source)], + ) diff --git a/tools/capture/screen_capture_selector.py b/tools/capture/screen_capture_selector.py new file mode 100644 index 0000000..0d7edea --- /dev/null +++ b/tools/capture/screen_capture_selector.py @@ -0,0 +1,345 @@ +"""Capability-level screen capture selector — routes between FFmpeg and Cap. + +Presents two options to the agent/user: + 1. FFmpeg (screen_recorder) — ready immediately, CLI-driven, no webcam + 2. Cap (cap_recorder) — needs install, polished UI, webcam overlay, cursor effects + +Provider discovery is automatic via the registry (capability="screen_capture"). +""" + +from __future__ import annotations + +from typing import Any + +from tools.base_tool import ( + BaseTool, + Determinism, + ExecutionMode, + ResourceProfile, + ToolResult, + ToolRuntime, + ToolStability, + ToolStatus, + ToolTier, +) + + +class ScreenCaptureSelector(BaseTool): + name = "screen_capture_selector" + version = "0.1.0" + tier = ToolTier.SOURCE + capability = "screen_capture" + provider = "selector" + stability = ToolStability.BETA + execution_mode = ExecutionMode.SYNC + determinism = Determinism.DETERMINISTIC + runtime = ToolRuntime.HYBRID + + agent_skills = ["screen-demo"] + + capabilities = [ + "screen_recording", + "provider_selection", + "cap_setup_guidance", + ] + + best_for = [ + "Choosing between quick FFmpeg recording and polished Cap recording", + "Guiding users through screen capture setup", + "Routing screen-demo pipeline to the right capture tool", + ] + + not_good_for = [ + "Direct screen recording (use the selected provider instead)", + ] + + input_schema = { + "type": "object", + "required": ["operation"], + "properties": { + "operation": { + "type": "string", + "enum": ["recommend", "record", "pick_latest"], + "description": ( + "'recommend' — assess available options and recommend one, " + "'record' — record screen using specified or best provider, " + "'pick_latest' — grab the most recent recording from any provider" + ), + }, + "preferred_provider": { + "type": "string", + "enum": ["auto", "ffmpeg", "cap"], + "default": "auto", + "description": "Provider preference. 'auto' picks the best available.", + }, + "output_path": { + "type": "string", + "description": "Path for the output MP4 file (required for 'record' operation)", + }, + "duration_seconds": { + "type": "integer", + "default": 60, + "description": "Recording duration in seconds (FFmpeg only)", + }, + "fps": { + "type": "integer", + "default": 30, + "description": "Frames per second (FFmpeg only)", + }, + "capture_audio": { + "type": "boolean", + "default": True, + "description": "Whether to capture audio (FFmpeg only)", + }, + "region": { + "type": "object", + "description": "Screen region to capture (FFmpeg only)", + "properties": { + "x": {"type": "integer"}, + "y": {"type": "integer"}, + "width": {"type": "integer"}, + "height": {"type": "integer"}, + }, + }, + "since_minutes": { + "type": "integer", + "default": 5, + "description": "For pick_latest: look back this many minutes", + }, + }, + } + + output_schema = { + "type": "object", + "properties": { + "recommended_provider": {"type": "string"}, + "options": {"type": "array"}, + "output_path": {"type": "string"}, + "capture_method": {"type": "string"}, + }, + } + + resource_profile = ResourceProfile( + cpu_cores=1, ram_mb=64, vram_mb=0, disk_mb=0, network_required=False, + ) + + side_effects = [] + + def _providers(self) -> dict[str, BaseTool]: + """Auto-discover screen_capture providers from the registry.""" + from tools.tool_registry import registry + registry.ensure_discovered() + tools = registry.get_by_capability("screen_capture") + return {t.provider: t for t in tools if t.name != self.name} + + @property + def fallback_tools(self) -> list[str]: + return list(self._providers().keys()) + + def get_status(self) -> ToolStatus: + providers = self._providers() + if any(t.get_status() == ToolStatus.AVAILABLE for t in providers.values()): + return ToolStatus.AVAILABLE + return ToolStatus.UNAVAILABLE + + def execute(self, inputs: dict[str, Any]) -> ToolResult: + operation = inputs["operation"] + + if operation == "recommend": + return self._recommend(inputs) + elif operation == "record": + return self._record(inputs) + elif operation == "pick_latest": + return self._pick_latest(inputs) + else: + return ToolResult( + success=False, + error=f"Unknown operation: {operation}. Valid: recommend, record, pick_latest", + ) + + def _recommend(self, inputs: dict[str, Any]) -> ToolResult: + """Assess both providers and return a recommendation with tradeoffs.""" + providers = self._providers() + + ffmpeg_tool = providers.get("ffmpeg") + cap_tool = providers.get("cap") + + options = [] + + # FFmpeg option — always available if ffmpeg is installed + if ffmpeg_tool: + ffmpeg_status = ffmpeg_tool.get_status() + options.append({ + "provider": "ffmpeg", + "tool": "screen_recorder", + "label": "Quick Recording (FFmpeg)", + "available": ffmpeg_status == ToolStatus.AVAILABLE, + "setup_required": ffmpeg_status != ToolStatus.AVAILABLE, + "strengths": [ + "Ready immediately — no additional install", + "CLI-driven — works in automated pipelines", + "Full screen or region capture", + "System + microphone audio", + ], + "limitations": [ + "No webcam overlay (picture-in-picture)", + "No cursor highlight or click effects", + "No built-in editor or captions", + "Raw capture — no polish", + ], + "best_when": "You need a quick recording or automated capture", + }) + + # Cap option — may need install + if cap_tool: + cap_detect = cap_tool.execute({"operation": "detect"}) + cap_installed = cap_detect.data.get("installed", False) if cap_detect.success else False + cap_running = cap_detect.data.get("running", False) if cap_detect.success else False + + status_label = "Running" if cap_running else ("Installed" if cap_installed else "Not installed") + options.append({ + "provider": "cap", + "tool": "cap_recorder", + "label": "Pro Recording (Cap)", + "available": cap_installed, + "running": cap_running, + "status": status_label, + "setup_required": not cap_installed, + "strengths": [ + "Webcam overlay (picture-in-picture)", + "Cursor highlight and click effects", + "GPU-accelerated capture", + "Built-in editor with auto-captions", + "Clean system audio capture", + "Polished, professional output", + ], + "limitations": [ + "Requires separate install (~2 min)", + "User must interact with Cap's UI to record", + "Cannot be fully automated from CLI", + ], + "best_when": "You want professional, polished screen recordings", + "setup_time": "~2 minutes" if not cap_installed else None, + }) + + # Determine recommendation + preferred = inputs.get("preferred_provider", "auto") + if preferred == "cap" and any(o["provider"] == "cap" for o in options): + recommended = "cap" + elif preferred == "ffmpeg" and any(o["provider"] == "ffmpeg" for o in options): + recommended = "ffmpeg" + else: + # Auto: recommend Cap if installed+running, otherwise FFmpeg + cap_option = next((o for o in options if o["provider"] == "cap"), None) + if cap_option and cap_option.get("running"): + recommended = "cap" + elif any(o["provider"] == "ffmpeg" and o["available"] for o in options): + recommended = "ffmpeg" + elif cap_option and cap_option.get("available"): + recommended = "cap" + else: + recommended = "ffmpeg" + + return ToolResult( + success=True, + data={ + "recommended_provider": recommended, + "options": options, + "message": self._build_recommendation_message(recommended, options), + }, + ) + + def _build_recommendation_message(self, recommended: str, options: list[dict]) -> str: + """Build a human-readable recommendation message for the agent to present.""" + cap_option = next((o for o in options if o["provider"] == "cap"), None) + ffmpeg_option = next((o for o in options if o["provider"] == "ffmpeg"), None) + + lines = ["**Screen Recording Options:**\n"] + + if ffmpeg_option: + status = "Ready" if ffmpeg_option["available"] else "Needs FFmpeg install" + lines.append(f"**Option 1 — Quick Recording (FFmpeg)** [{status}]") + lines.append(" Basic screen capture, works immediately. No webcam or effects.\n") + + if cap_option: + status = cap_option.get("status", "Unknown") + lines.append(f"**Option 2 — Pro Recording (Cap)** [{status}]") + lines.append(" Webcam overlay, cursor effects, built-in editor. Professional output.") + if not cap_option.get("available"): + lines.append(" Setup takes ~2 minutes. I can guide you through it.\n") + else: + lines.append("") + + lines.append(f"**Recommended:** {recommended}") + return "\n".join(lines) + + def _record(self, inputs: dict[str, Any]) -> ToolResult: + """Route a record request to the appropriate provider.""" + preferred = inputs.get("preferred_provider", "auto") + providers = self._providers() + + # Determine which provider to use + if preferred == "cap": + tool = providers.get("cap") + if tool: + # Cap doesn't do the actual recording — it picks up what Cap recorded + return tool.execute({"operation": "pick_latest", "output_dir": inputs.get("output_path")}) + return ToolResult(success=False, error="Cap provider not found in registry.") + + if preferred == "ffmpeg" or preferred == "auto": + tool = providers.get("ffmpeg") + if tool and tool.get_status() == ToolStatus.AVAILABLE: + return tool.execute({ + "output_path": inputs.get("output_path", "recording.mp4"), + "duration_seconds": inputs.get("duration_seconds", 60), + "fps": inputs.get("fps", 30), + "capture_audio": inputs.get("capture_audio", True), + "region": inputs.get("region"), + }) + + # FFmpeg not available — try Cap + cap_tool = providers.get("cap") + if cap_tool: + cap_detect = cap_tool.execute({"operation": "detect"}) + if cap_detect.success and cap_detect.data.get("running"): + return cap_tool.execute({ + "operation": "pick_latest", + "output_dir": inputs.get("output_path"), + }) + + return ToolResult( + success=False, + error="No screen capture provider available. Install FFmpeg or Cap.", + ) + + return ToolResult(success=False, error=f"Unknown provider: {preferred}") + + def _pick_latest(self, inputs: dict[str, Any]) -> ToolResult: + """Try to pick the latest recording from any available provider.""" + providers = self._providers() + since = inputs.get("since_minutes", 5) + + # Try Cap first (more likely to have user-initiated recordings) + cap_tool = providers.get("cap") + if cap_tool: + result = cap_tool.execute({ + "operation": "find_recordings", + "since_minutes": since, + }) + if result.success and result.data.get("recordings"): + latest = result.data["recordings"][0] + return ToolResult( + success=True, + data={ + "output_path": latest["path"], + "size_mb": latest["size_mb"], + "capture_method": "cap", + "source": "cap_recordings_dir", + }, + artifacts=[latest["path"]], + ) + + return ToolResult( + success=False, + error="No recent recordings found. Record something first using Cap or FFmpeg.", + ) diff --git a/tools/capture/screen_recorder.py b/tools/capture/screen_recorder.py new file mode 100644 index 0000000..8bf7ece --- /dev/null +++ b/tools/capture/screen_recorder.py @@ -0,0 +1,394 @@ +"""FFmpeg-based screen recorder. + +Cross-platform screen capture using FFmpeg's native capture devices. +Records screen + optional audio to MP4. Designed as the "quick start" +option — no install beyond FFmpeg, works everywhere, CLI-driven. + +Platform capture devices: + Windows: gdigrab (screen) + dshow (audio) + macOS: avfoundation (screen + audio) + Linux: x11grab (screen) + pulse (audio) +""" + +from __future__ import annotations + +import os +import platform +import signal +import subprocess +import time +from pathlib import Path +from typing import Any + +from tools.base_tool import ( + BaseTool, + Determinism, + ExecutionMode, + ResourceProfile, + ToolResult, + ToolRuntime, + ToolStability, + ToolTier, +) + + +def _detect_audio_device_windows() -> str | None: + """Find a working audio input device on Windows via dshow.""" + try: + result = subprocess.run( + ["ffmpeg", "-list_devices", "true", "-f", "dshow", "-i", "dummy"], + capture_output=True, text=True, timeout=10, + ) + # dshow lists devices in stderr + output = result.stderr + lines = output.splitlines() + for i, line in enumerate(lines): + if "audio" in line.lower() and "DirectShow audio" in line: + # Next line(s) contain actual device names + for j in range(i + 1, min(i + 10, len(lines))): + if '"' in lines[j] and "Alternative name" not in lines[j]: + name = lines[j].split('"')[1] + return name + except Exception: + pass + return None + + +def _detect_audio_device_mac() -> str | None: + """Find the default audio input index on macOS via avfoundation.""" + try: + result = subprocess.run( + ["ffmpeg", "-f", "avfoundation", "-list_devices", "true", "-i", ""], + capture_output=True, text=True, timeout=10, + ) + output = result.stderr + in_audio = False + for line in output.splitlines(): + if "AVFoundation audio devices" in line: + in_audio = True + continue + if in_audio and "[" in line and "]" in line: + # Return first audio device index + idx = line.split("[")[1].split("]")[0].strip() + if idx.isdigit(): + return idx + except Exception: + pass + return None + + +class ScreenRecorder(BaseTool): + name = "screen_recorder" + version = "0.1.0" + tier = ToolTier.SOURCE + capability = "screen_capture" + provider = "ffmpeg" + stability = ToolStability.BETA + execution_mode = ExecutionMode.SYNC + determinism = Determinism.DETERMINISTIC + runtime = ToolRuntime.LOCAL + + dependencies = ["binary:ffmpeg"] + install_instructions = ( + "Install ffmpeg:\n" + " Windows: winget install ffmpeg\n" + " macOS: brew install ffmpeg\n" + " Linux: sudo apt install ffmpeg" + ) + + capabilities = [ + "record_screen", + "record_screen_with_audio", + "record_region", + ] + + best_for = [ + "Quick screen recording without additional software", + "Automated screen capture for demo pipelines", + "Recording specific screen regions for tutorials", + ] + + not_good_for = [ + "Webcam overlay (PiP) — use Cap for that", + "Cursor highlight effects — use Cap for that", + "Interactive recording with pause/resume UI", + ] + + input_schema = { + "type": "object", + "required": ["output_path"], + "properties": { + "output_path": { + "type": "string", + "description": "Path for the output MP4 file", + }, + "duration_seconds": { + "type": "integer", + "default": 60, + "description": "Recording duration in seconds (default: 60, max: 600)", + }, + "fps": { + "type": "integer", + "default": 30, + "description": "Frames per second (15, 24, 30, or 60)", + }, + "capture_audio": { + "type": "boolean", + "default": True, + "description": "Whether to capture system/microphone audio", + }, + "region": { + "type": "object", + "description": "Optional screen region to capture (full screen if omitted)", + "properties": { + "x": {"type": "integer", "description": "Left offset in pixels"}, + "y": {"type": "integer", "description": "Top offset in pixels"}, + "width": {"type": "integer", "description": "Width in pixels"}, + "height": {"type": "integer", "description": "Height in pixels"}, + }, + }, + "screen_index": { + "type": "integer", + "default": 0, + "description": "Monitor index for multi-monitor setups (0 = primary)", + }, + }, + } + + output_schema = { + "type": "object", + "properties": { + "output_path": {"type": "string"}, + "duration_seconds": {"type": "number"}, + "resolution": {"type": "string"}, + "has_audio": {"type": "boolean"}, + "file_size_mb": {"type": "number"}, + }, + } + + resource_profile = ResourceProfile( + cpu_cores=2, ram_mb=512, vram_mb=0, disk_mb=500, network_required=False, + ) + + side_effects = ["creates_file"] + fallback_tools = ["cap_recorder"] + + def execute(self, inputs: dict[str, Any]) -> ToolResult: + output_path = Path(inputs["output_path"]) + duration = min(inputs.get("duration_seconds", 60), 600) + fps = inputs.get("fps", 30) + capture_audio = inputs.get("capture_audio", True) + region = inputs.get("region") + screen_index = inputs.get("screen_index", 0) + + output_path.parent.mkdir(parents=True, exist_ok=True) + + sys_platform = platform.system() + cmd = self._build_command( + sys_platform, str(output_path), duration, fps, + capture_audio, region, screen_index, + ) + + if cmd is None: + return ToolResult( + success=False, + error=f"Screen recording not supported on {sys_platform}. " + f"Supported: Windows, macOS, Linux.", + ) + + start_time = time.time() + try: + proc = subprocess.run( + cmd, capture_output=True, text=True, + timeout=duration + 30, # grace period + ) + elapsed = time.time() - start_time + + if not output_path.exists(): + return ToolResult( + success=False, + error=f"Recording failed — no output file. FFmpeg stderr: {proc.stderr[-500:]}", + ) + + file_size_mb = output_path.stat().st_size / (1024 * 1024) + + # Probe the output to get actual resolution + resolution = self._probe_resolution(str(output_path)) + + return ToolResult( + success=True, + data={ + "output_path": str(output_path), + "duration_seconds": round(elapsed, 1), + "resolution": resolution, + "has_audio": capture_audio, + "file_size_mb": round(file_size_mb, 1), + "platform": sys_platform, + "capture_method": "ffmpeg", + }, + artifacts=[str(output_path)], + duration_seconds=elapsed, + ) + + except subprocess.TimeoutExpired: + # Recording completed by timeout — this is expected behavior + if output_path.exists(): + file_size_mb = output_path.stat().st_size / (1024 * 1024) + return ToolResult( + success=True, + data={ + "output_path": str(output_path), + "duration_seconds": duration, + "has_audio": capture_audio, + "file_size_mb": round(file_size_mb, 1), + "platform": sys_platform, + "capture_method": "ffmpeg", + }, + artifacts=[str(output_path)], + duration_seconds=duration, + ) + return ToolResult(success=False, error="Recording timed out with no output") + + except Exception as exc: + return ToolResult(success=False, error=str(exc)) + + def _build_command( + self, + sys_platform: str, + output_path: str, + duration: int, + fps: int, + capture_audio: bool, + region: dict | None, + screen_index: int, + ) -> list[str] | None: + """Build platform-specific FFmpeg capture command.""" + + if sys_platform == "Windows": + return self._build_windows_cmd( + output_path, duration, fps, capture_audio, region, + ) + elif sys_platform == "Darwin": + return self._build_mac_cmd( + output_path, duration, fps, capture_audio, region, screen_index, + ) + elif sys_platform == "Linux": + return self._build_linux_cmd( + output_path, duration, fps, capture_audio, region, + ) + return None + + def _build_windows_cmd( + self, output_path: str, duration: int, fps: int, + capture_audio: bool, region: dict | None, + ) -> list[str]: + cmd = ["ffmpeg", "-y"] + + # Video input: gdigrab + cmd += ["-f", "gdigrab"] + cmd += ["-framerate", str(fps)] + cmd += ["-t", str(duration)] + + if region: + cmd += ["-offset_x", str(region.get("x", 0))] + cmd += ["-offset_y", str(region.get("y", 0))] + cmd += ["-video_size", f"{region['width']}x{region['height']}"] + + cmd += ["-i", "desktop"] + + # Audio input: dshow + if capture_audio: + audio_device = _detect_audio_device_windows() + if audio_device: + cmd += ["-f", "dshow", "-i", f"audio={audio_device}"] + + # Output encoding + cmd += ["-c:v", "libx264", "-preset", "ultrafast", "-crf", "23"] + if capture_audio: + cmd += ["-c:a", "aac", "-b:a", "128k"] + cmd += ["-pix_fmt", "yuv420p"] + cmd += [output_path] + + return cmd + + def _build_mac_cmd( + self, output_path: str, duration: int, fps: int, + capture_audio: bool, region: dict | None, screen_index: int, + ) -> list[str]: + cmd = ["ffmpeg", "-y"] + + # avfoundation: "screen_index:audio_index" or "screen_index:none" + audio_idx = "none" + if capture_audio: + detected = _detect_audio_device_mac() + if detected: + audio_idx = detected + + cmd += ["-f", "avfoundation"] + cmd += ["-framerate", str(fps)] + cmd += ["-t", str(duration)] + + if region: + # avfoundation doesn't support region directly — we crop in post + cmd += ["-i", f"{screen_index}:{audio_idx}"] + cmd += ["-vf", f"crop={region['width']}:{region['height']}:{region.get('x', 0)}:{region.get('y', 0)}"] + else: + cmd += ["-i", f"{screen_index}:{audio_idx}"] + + cmd += ["-c:v", "libx264", "-preset", "ultrafast", "-crf", "23"] + if capture_audio and audio_idx != "none": + cmd += ["-c:a", "aac", "-b:a", "128k"] + cmd += ["-pix_fmt", "yuv420p"] + cmd += [output_path] + + return cmd + + def _build_linux_cmd( + self, output_path: str, duration: int, fps: int, + capture_audio: bool, region: dict | None, + ) -> list[str]: + cmd = ["ffmpeg", "-y"] + + # x11grab + display = os.environ.get("DISPLAY", ":0.0") + cmd += ["-f", "x11grab"] + cmd += ["-framerate", str(fps)] + cmd += ["-t", str(duration)] + + if region: + cmd += ["-video_size", f"{region['width']}x{region['height']}"] + cmd += ["-i", f"{display}+{region.get('x', 0)},{region.get('y', 0)}"] + else: + # Full screen — need to detect resolution + cmd += ["-i", display] + + if capture_audio: + cmd += ["-f", "pulse", "-i", "default"] + + cmd += ["-c:v", "libx264", "-preset", "ultrafast", "-crf", "23"] + if capture_audio: + cmd += ["-c:a", "aac", "-b:a", "128k"] + cmd += ["-pix_fmt", "yuv420p"] + cmd += [output_path] + + return cmd + + def _probe_resolution(self, path: str) -> str: + """Get video resolution via ffprobe.""" + try: + result = subprocess.run( + [ + "ffprobe", "-v", "error", + "-select_streams", "v:0", + "-show_entries", "stream=width,height", + "-of", "csv=p=0", + path, + ], + capture_output=True, text=True, timeout=10, + ) + parts = result.stdout.strip().split(",") + if len(parts) == 2: + return f"{parts[0]}x{parts[1]}" + except Exception: + pass + return "unknown"