Add screen capture tools with FFmpeg and Cap dual-provider system
New capture layer for the screen-demo pipeline: FFmpeg for quick CLI-driven recording, Cap integration for polished recordings with webcam overlay and cursor effects. Selector presents both options and routes based on availability.
This commit is contained in:
calesthio
@@ -0,0 +1 @@
|
||||
"""Screen capture tools — FFmpeg native recording and Cap integration."""
|
||||
@@ -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)],
|
||||
)
|
||||
@@ -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.",
|
||||
)
|
||||
@@ -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"
|
||||
Reference in New Issue
Block a user