diff --git a/.agents/skills/zapcap-captions/SKILL.md b/.agents/skills/zapcap-captions/SKILL.md new file mode 100644 index 000000000..966c44633 --- /dev/null +++ b/.agents/skills/zapcap-captions/SKILL.md @@ -0,0 +1,159 @@ +--- +name: zapcap-captions +description: Add animated, word-synced styled captions to a video with the ZapCap API. Use when the user wants viral/social caption styles (Hormozi, Beast, Devin, ...), word-level highlight captions burned into a video, emoji captions, or translated+captioned variants of the same video. Backs the OpenMontage `zapcap_captions` subtitle provider. +metadata: + openclaw: + requires: + env: + - ZAPCAP_API_KEY + primaryEnv: ZAPCAP_API_KEY +--- + +# ZapCap Captions + +ZapCap is a captioning API. You upload a video, choose a styled caption +*template*, and ZapCap transcribes the audio and renders animated, word-synced +subtitles burned into the video — the fastest path to social-ready captions +(TikTok/Reels/Shorts). + +This skill backs the `zapcap_captions` tool (`tools/subtitle/zapcap_captions.py`), +capability `subtitle`, provider `zapcap`. + +## When to use it (vs. the local subtitle tools) + +| Need | Tool | +|------|------| +| Viral styled, animated, word-highlight captions burned in | **`zapcap_captions`** (this) | +| Word-by-word caption burn locally via Remotion | `remotion_caption_burn` | +| Plain SRT/VTT/caption JSON from existing transcript segments | `subtitle_gen` | + +ZapCap needs an **audio track with speech** — it transcribes to make captions. +Don't use it for silent/music-only videos. Max video length is **30 minutes**. + +## Configuration + +```bash +ZAPCAP_API_KEY=... # required (sent as the x-api-key header) +``` + +Calls go to `https://api.zapcap.ai`. This matches the `zapcap-mcp` server's env +var, so one `.env` configures both the OpenMontage tool and the MCP. + +## The flow (what the tool does on `action="caption"`) + +1. **Upload** — `POST /videos` (local file, multipart `file`) or + `POST /videos/url` (`{url}`) → `videoId`. +2. **Create task** — `POST /videos/{videoId}/task` with `templateId` + (+ options) → `taskId`. Set `autoApprove: true` so rendering starts without + a manual transcript-approval step (the tool defaults to this). +3. **Poll** — `GET /videos/{videoId}/task/{taskId}` until `status` is + `completed`. Statuses: `pending → transcribing → transcriptionCompleted → + rendering → completed` (or `failed`). +4. **Download** — stream `downloadUrl` from the completed task to `output_path`. + +Auth on every call: header `x-api-key: $ZAPCAP_API_KEY`. + +## Templates + +List them first (names are not stable IDs — always resolve): + +```python +from tools.subtitle.zapcap_captions import ZapCapCaptions +tpl = ZapCapCaptions().execute({"action": "list_templates"}) +# tpl.data["templates"] -> [{id, name, categories}, ...] +``` + +Popular templates (categories: `animated`, `highlighted`, `basic`): +`Hormozi 1` (animated+highlighted), `Beast`, `Devin`, `Ella`, `Tracy` (basic), +`Karl`, `Maya`, `Hormozi 2/3/4/5`. The tool accepts either `template_id` +(UUID) or `template_name` (resolved case-insensitively via `/templates`). + +## OpenMontage usage + +Full caption of a local file (the common case): + +```python +from tools.subtitle.zapcap_captions import ZapCapCaptions + +res = ZapCapCaptions().execute({ + "action": "caption", # default; can omit + "input_path": "projects/demo/renders/final.mp4", + "template_name": "Hormozi 1", # or template_id="a51c5222-..." + "language": "en", # omit to auto-detect + "auto_approve": True, + "output_path": "projects/demo/renders/final_captioned.mp4", + "timeout_seconds": 600, +}) +# res.data -> {videoId, taskId, templateId, downloadUrl, transcriptUrl, output, ...} +# res.artifacts -> ["projects/demo/renders/final_captioned.mp4"] +``` + +Caption a public URL instead: pass `video_url` instead of `input_path`. + +## renderOptions (subtitle appearance) + +Pass any of these under `render_options`; all optional: + +```python +"render_options": { + "subsOptions": { + "emoji": True, "emojiAnimation": True, + "emphasizeKeywords": True, # highlight key words per template style + "animation": True, "punctuation": False, + "displayWords": 4, # words shown at once (guidance) + }, + "styleOptions": { + "top": 40, # Y position, % of height (higher = lower) + "fontUppercase": True, "fontSize": 46, "fontWeight": 900, + "fontColor": "#ffffff", + "fontShadow": "l", # none|s|m|l + "stroke": "s", "strokeColor": "#000000", + "fontId": "", # optional, from POST /fonts + }, + "highlightOptions": { # colors used for emphasized keywords + "randomColourOne": "#2bf82a", + "randomColourTwo": "#fdfa14", + "randomColourThree": "#f01916", + }, +} +``` + +## Translation & fan-out (transcribe once, render many) + +To caption one video into several languages or templates **without paying to +transcribe N times**, transcribe once and reuse the transcript: + +```python +tool = ZapCapCaptions() +# 1. upload + create first task, but only wait for transcription +up = tool.execute({"action": "upload", "input_path": "in.mp4"}) +vid = up.data["videoId"] +first = tool.execute({"action": "create_task", "video_id": vid, + "template_name": "Hormozi 1", "language": "en"}) +# (poll first.data["taskId"] to transcriptionCompleted with get_task) + +# 2. fan out: reuse the transcript, translate per target +for lang in ["es", "fr", "de"]: + tool.execute({"action": "create_task", "video_id": vid, + "template_name": "Hormozi 1", + "transcript_task_id": first.data["taskId"], + "translate_to": lang, "auto_approve": True}) +``` + +The `mcp__zapcap__*` MCP tools (`upload_video_*`, `create_video_task`, +`wait_for_task`) expose the same flow and `wait_for_task` supports +`waitFor="transcriptionCompleted"` for the fan-out barrier. + +## Pipelines that benefit + +`clip-factory`, `podcast-repurpose`, `talking-head`, and `localization-dub` +all produce speech-driven social clips where styled burned-in captions are the +norm. Offer `zapcap_captions` at the edit/compose stage for those briefs as an +alternative to `remotion_caption_burn` when the user wants ZapCap's template +styles. + +## Failure modes + +- `401/403` → bad or missing `ZAPCAP_API_KEY`, or no API credits / wrong plan. +- task `status: failed` with an `error` → usually no speech, unsupported codec, + or >30 min. Re-check the source has an audio track. diff --git a/.env.example b/.env.example index cb32576a3..3dbd98512 100644 --- a/.env.example +++ b/.env.example @@ -32,6 +32,10 @@ PEXELS_API_KEY= # Pexels stock footage/images (free) PIXABAY_API_KEY= # Pixabay stock footage/images (free) UNSPLASH_ACCESS_KEY= # Unsplash stock images (free developer key) +# --- Captions (styled subtitles) --- +ZAPCAP_API_KEY= # ZapCap captioning API — animated word-synced caption templates (Hormozi, Beast, ...) + # Get one at https://platform.zapcap.ai/dashboard/api-key (Pro plan + credits) + # --- Analysis --- HF_TOKEN= # HuggingFace token — enables speaker diarization in transcriber diff --git a/AGENT_GUIDE.md b/AGENT_GUIDE.md index cf7439760..07d932461 100644 --- a/AGENT_GUIDE.md +++ b/AGENT_GUIDE.md @@ -643,6 +643,7 @@ The `.agents/skills/` directory is large. When you're not coming in through a to | **Image generation** | `bfl-api`, `flux-best-practices` | | **Video generation** | `seedance-2-0` (preferred premium default — cinematic, trailer, multi-shot, synced audio, lip-sync), `ai-video-gen`, `ltx2` | | **Audio** | `elevenlabs`, `music`, `sound-effects`, `acestep`, `text-to-speech`, `setup-api-key` | +| **Captions / subtitles** | `zapcap-captions` (styled/animated word-synced caption templates via the `zapcap_captions` tool) | | **Avatar / lip-sync** | `avatar-video`, `heygen`, `create-video`, `faceswap`, `video-translate`, `speech-to-text`, `agents` | | **Capture** | `playwright-recording` (browser flows), `ffmpeg` (post) | | **Visualization** | `beautiful-mermaid`, `d3-viz`, `manim-composer`, `manimce-best-practices`, `manimgl-best-practices` | diff --git a/README.md b/README.md index 0d24608ad..83f50e16d 100644 --- a/README.md +++ b/README.md @@ -207,7 +207,7 @@ You don't need paid API keys to make real videos. Out of the box, `make setup` g | **Composition (React)** | Remotion | React-based rendering — spring-animated image scenes, text cards, stat cards, charts, TikTok-style word-level captions, TalkingHead | | **Composition (HTML/GSAP)** | HyperFrames | HTML/CSS/GSAP rendering — kinetic typography, product promos, launch reels, registry blocks, website-to-video, rigged SVG character animation | | **Post-production** | FFmpeg | Encoding, subtitle burn-in, audio mixing, color grading | -| **Subtitles** | Built-in | Auto-generated captions with word-level timing | +| **Subtitles** | Built-in + ZapCap | Auto-generated captions with word-level timing; optional styled/animated caption templates (Hormozi, Beast, ...) via ZapCap | OpenMontage picks between Remotion and HyperFrames at proposal time (locked as `render_runtime`). Remotion is the default for data-driven explainers and anything using the existing React scene stack; HyperFrames is the default for motion-graphics-heavy briefs that express naturally as HTML + GSAP, including the `character-animation` pipeline's SVG/GSAP rig output. See `skills/core/hyperframes.md` for the full decision matrix. diff --git a/docs/PROVIDERS.md b/docs/PROVIDERS.md index f0460a69e..f033c9d34 100644 --- a/docs/PROVIDERS.md +++ b/docs/PROVIDERS.md @@ -20,6 +20,7 @@ Everything you need to know about every provider in OpenMontage — setup instru | 8 | **$12/month** | Runway | Gen-4 video — highest quality AI video | | 9 | **pay-as-you-go** | HeyGen | Avatar videos, multi-model video gateway | | 10 | **pay-as-you-go** | Suno | Full song generation with vocals and lyrics | +| 10b | **Pro plan + credits** | ZapCap | Animated/word-synced styled captions (Hormozi, Beast, ...) | | 11 | **$0 + GPU** | Local video gen | WAN 2.1, Hunyuan, CogVideo, LTX — free, offline | | 12 | **$0 + GPU** | Local Diffusion | Stable Diffusion images — free, offline | @@ -50,6 +51,9 @@ HEYGEN_API_KEY= # HeyGen avatar video gateway RUNWAY_API_KEY= # Runway Gen-4 video (direct) SUNO_API_KEY= # Suno music generation +# CAPTIONS (styled subtitles) +ZAPCAP_API_KEY= # ZapCap animated/word-synced caption templates + # LOCAL (no keys needed — just GPU + install) VIDEO_GEN_LOCAL_ENABLED= # Set to "true" for local video gen VIDEO_GEN_LOCAL_MODEL= # wan2.1-1.3b, wan2.1-14b, hunyuan-1.5, ltx2-local, cogvideo-5b @@ -207,6 +211,56 @@ Doubao Speech 2.0 is billed by character package or usage in Volcengine. OpenMon --- +### ZapCap — Caption Templates + +> **Animated, word-synced caption templates.** Upload a video, pick a template (Hormozi, Beast, Devin, ...), and ZapCap transcribes and burns in styled subtitles. + +**Tools unlocked:** `zapcap_captions` +**Env var:** `ZAPCAP_API_KEY` +**Layer 3 skill:** `.agents/skills/zapcap-captions/SKILL.md` + +#### Setup + +1. Subscribe to a ZapCap Pro plan and buy API credits at [platform.zapcap.ai](https://platform.zapcap.ai/dashboard/billing). +2. Copy your key from the [API key page](https://platform.zapcap.ai/dashboard/api-key). +3. Add to `.env`: `ZAPCAP_API_KEY=your-key-here` + +#### What It Is Best For + +- Viral caption styles (Hormozi, Beast, Devin, Ella, ...) out of the box +- Word-level highlight + emoji captions burned into the video +- Translating + captioning the same video into many languages (transcribe once, fan out) +- Speech-driven social clips (`clip-factory`, `podcast-repurpose`, `talking-head`, `localization-dub`) + +Needs a speech audio track (it transcribes to caption). Max length 30 min. +For local/offline burn-in use `remotion_caption_burn`; for plain SRT/VTT use +`subtitle_gen`. + +#### Usage + +```python +from tools.subtitle.zapcap_captions import ZapCapCaptions + +res = ZapCapCaptions().execute({ + "input_path": "projects/demo/renders/final.mp4", + "template_name": "Hormozi 1", # or template_id="a51c5222-..." + "output_path": "projects/demo/renders/final_captioned.mp4", +}) +# res.artifacts -> ["projects/demo/renders/final_captioned.mp4"] +``` + +`action="list_templates"` lists templates; `render_options` overrides subtitle +appearance; `translate_to` + `transcript_task_id` drive multi-language fan-out. +See the Layer 3 skill for the full schema. + +#### Pricing + +Billed in ZapCap API credits (Pro plan + credits required). Cost scales with +video minutes processed. The tool does not estimate USD; track credits on the +[ZapCap dashboard](https://platform.zapcap.ai/dashboard/billing). + +--- + ### Google — TTS + Imagen (Shared Key) > **One key, two tools.** Google Cloud TTS has 700+ voices in 50+ languages — the strongest localization option. Imagen 4 generates high-quality images. diff --git a/tests/tools/test_zapcap_captions.py b/tests/tools/test_zapcap_captions.py new file mode 100644 index 000000000..6f44443f2 --- /dev/null +++ b/tests/tools/test_zapcap_captions.py @@ -0,0 +1,276 @@ +"""Offline contract tests for the ZapCap captions provider. + +All network I/O is mocked at the tool's ``_request`` / ``_download`` boundary — +these tests never touch api.zapcap.ai and require no API key. They cover: + +- registry discovery under capability="subtitle" +- get_status() with no key / missing requests / configured key +- local missing-file failure for input_path +- template-name resolution and not-found behavior +- mocked happy path: upload -> create_task -> poll -> download +- a mocked API error path with useful, non-leaky error reporting +""" + +from __future__ import annotations + +import builtins + +import pytest + +from tools.base_tool import ToolStatus +from tools.subtitle import zapcap_captions as zc +from tools.subtitle.zapcap_captions import BASE_URL, ZapCapCaptions +from tools.tool_registry import registry + +_TEST_KEY = "test-key-do-not-log-1234567890" + +# Canned /templates response used across resolution + happy-path tests. +_TEMPLATES = [ + {"id": "a51c5222-47a7-4c37-b052-7b9853d66bf6", "name": "Hormozi 1", "categories": ["animated"]}, + {"id": "46d20d67-255c-4c6a-b971-31fddcfea7f0", "name": "Beast", "categories": ["highlighted"]}, +] + + +def _tool(monkeypatch, *, key: str | None = _TEST_KEY) -> ZapCapCaptions: + """A ZapCapCaptions instance with a deterministic env key.""" + if key is None: + monkeypatch.delenv("ZAPCAP_API_KEY", raising=False) + else: + monkeypatch.setenv("ZAPCAP_API_KEY", key) + return ZapCapCaptions() + + +# --------------------------------------------------------------------------- # +# Registry discovery + contract metadata +# --------------------------------------------------------------------------- # + +def test_registry_discovers_zapcap_under_subtitle(): + registry.discover() + subtitle_tools = {t.name for t in registry.get_by_capability("subtitle")} + assert "zapcap_captions" in subtitle_tools + + tool = registry.get("zapcap_captions") + assert tool is not None + assert tool.capability == "subtitle" + assert tool.provider == "zapcap" + assert tool.runtime.value == "api" + # Declares the required key as a dependency and links its Layer 3 skill. + assert "env:ZAPCAP_API_KEY" in tool.dependencies + assert "zapcap-captions" in tool.agent_skills + + +def test_base_url_is_production(): + assert BASE_URL == "https://api.zapcap.ai" + + +# --------------------------------------------------------------------------- # +# get_status() +# --------------------------------------------------------------------------- # + +def test_status_unavailable_without_key(monkeypatch): + tool = _tool(monkeypatch, key=None) + assert tool.get_status() == ToolStatus.UNAVAILABLE + + +def test_status_unavailable_when_requests_missing(monkeypatch): + tool = _tool(monkeypatch) # key present + real_import = builtins.__import__ + + def fake_import(name, *args, **kwargs): + if name == "requests": + raise ImportError("simulated: requests not installed") + return real_import(name, *args, **kwargs) + + monkeypatch.setattr(builtins, "__import__", fake_import) + assert tool.get_status() == ToolStatus.UNAVAILABLE + + +def test_status_available_with_key_and_requests(monkeypatch): + tool = _tool(monkeypatch) + assert tool.get_status() == ToolStatus.AVAILABLE + + +def test_execute_without_key_returns_error(monkeypatch): + tool = _tool(monkeypatch, key=None) + result = tool.execute({"action": "list_templates"}) + assert result.success is False + assert "ZAPCAP_API_KEY" in result.error + + +# --------------------------------------------------------------------------- # +# Local missing-file failure +# --------------------------------------------------------------------------- # + +def test_caption_missing_input_file_fails_cleanly(monkeypatch, tmp_path): + tool = _tool(monkeypatch) + missing = tmp_path / "nope.mp4" + + # Guard: no network call should ever happen for a missing local file. + def boom(*args, **kwargs): # pragma: no cover - must not be reached + raise AssertionError("network was called for a missing local file") + + monkeypatch.setattr(tool, "_request", boom) + + result = tool.execute( + {"action": "caption", "input_path": str(missing), "template_id": _TEMPLATES[0]["id"]} + ) + assert result.success is False + assert "not found" in result.error.lower() + assert str(missing) in result.error + + +# --------------------------------------------------------------------------- # +# Template-name resolution +# --------------------------------------------------------------------------- # + +def test_resolve_template_id_by_name_case_insensitive(monkeypatch): + tool = _tool(monkeypatch) + monkeypatch.setattr(tool, "_list_templates", lambda: _TEMPLATES) + # Case-insensitive name match returns the id. + assert tool._resolve_template_id(None, "hormozi 1") == _TEMPLATES[0]["id"] + # Explicit id passes through untouched (no lookup needed). + assert tool._resolve_template_id("explicit-id", None) == "explicit-id" + + +def test_resolve_template_id_unknown_name_raises_with_options(monkeypatch): + tool = _tool(monkeypatch) + monkeypatch.setattr(tool, "_list_templates", lambda: _TEMPLATES) + with pytest.raises(ValueError) as exc: + tool._resolve_template_id(None, "Nonexistent Template") + msg = str(exc.value) + assert "Nonexistent Template" in msg + # The error lists the available names to guide the caller. + assert "Hormozi 1" in msg and "Beast" in msg + + +def test_resolve_template_id_requires_id_or_name(monkeypatch): + tool = _tool(monkeypatch) + with pytest.raises(ValueError): + tool._resolve_template_id(None, None) + + +# --------------------------------------------------------------------------- # +# Mocked happy path: upload -> create_task -> poll -> download +# --------------------------------------------------------------------------- # + +def test_caption_happy_path(monkeypatch, tmp_path): + tool = _tool(monkeypatch) + + src = tmp_path / "in.mp4" + src.write_bytes(b"\x00\x00fake-mp4-bytes") + out = tmp_path / "out" / "captioned.mp4" + + # Task goes transcribing -> completed across two polls. + task_states = [ + {"id": "task-abc", "status": "transcribing", "downloadUrl": None, "transcript": None}, + { + "id": "task-abc", + "status": "completed", + "downloadUrl": "https://cdn.example/out.mp4", + "transcript": "https://cdn.example/words.json", + }, + ] + calls: list[tuple[str, str]] = [] + + def fake_request(method, path, *, json_body=None, params=None, files=None, timeout=120): + calls.append((method, path)) + if method == "POST" and path == "/videos": + return {"id": "vid-123", "status": "uploaded"} + if method == "POST" and path == "/videos/vid-123/task": + assert json_body["templateId"] == _TEMPLATES[0]["id"] + assert json_body["autoApprove"] is True # default + return {"taskId": "task-abc"} + if method == "GET" and path == "/videos/vid-123/task/task-abc": + return task_states.pop(0) if len(task_states) > 1 else task_states[0] + raise AssertionError(f"unexpected request: {method} {path}") + + downloaded: dict[str, str] = {} + + def fake_download(url, output_path): + downloaded["url"] = url + from pathlib import Path + + Path(output_path).parent.mkdir(parents=True, exist_ok=True) + Path(output_path).write_bytes(b"captioned-bytes") + return output_path + + monkeypatch.setattr(tool, "_request", fake_request) + monkeypatch.setattr(tool, "_download", fake_download) + + result = tool.execute( + { + "action": "caption", + "input_path": str(src), + "template_id": _TEMPLATES[0]["id"], + "output_path": str(out), + "poll_interval_seconds": 0, # no real sleeping + "timeout_seconds": 30, + } + ) + + assert result.success is True, result.error + assert result.data["videoId"] == "vid-123" + assert result.data["taskId"] == "task-abc" + assert result.data["templateId"] == _TEMPLATES[0]["id"] + assert result.data["status"] == "completed" + assert result.data["output"] == str(out) + assert result.artifacts == [str(out)] + assert out.exists() and out.read_bytes() == b"captioned-bytes" + assert downloaded["url"] == "https://cdn.example/out.mp4" + # Confirms the full lifecycle ran: upload, task, and at least two polls. + assert ("POST", "/videos") in calls + assert ("POST", "/videos/vid-123/task") in calls + assert calls.count(("GET", "/videos/vid-123/task/task-abc")) >= 2 + + +# --------------------------------------------------------------------------- # +# Mocked API error path — useful message, no secret leak +# --------------------------------------------------------------------------- # + +def test_caption_api_error_is_reported_without_leaking_key(monkeypatch, tmp_path): + tool = _tool(monkeypatch) + src = tmp_path / "in.mp4" + src.write_bytes(b"fake") + + def failing_request(method, path, *, json_body=None, params=None, files=None, timeout=120): + # Mirrors how _request surfaces a non-2xx response (body message only). + raise RuntimeError("ZapCap POST /videos -> 401 Unauthorized: invalid api key") + + monkeypatch.setattr(tool, "_request", failing_request) + + result = tool.execute( + {"action": "caption", "input_path": str(src), "template_id": _TEMPLATES[0]["id"]} + ) + assert result.success is False + # Error is actionable (names the action + upstream status). + assert "caption" in result.error + assert "401" in result.error + # The API key must never appear in surfaced errors. + assert _TEST_KEY not in result.error + + +def test_task_failed_status_surfaces_error(monkeypatch, tmp_path): + tool = _tool(monkeypatch) + src = tmp_path / "in.mp4" + src.write_bytes(b"fake") + + def fake_request(method, path, *, json_body=None, params=None, files=None, timeout=120): + if method == "POST" and path == "/videos": + return {"id": "vid-9", "status": "uploaded"} + if path.endswith("/task"): + return {"taskId": "task-9"} + # GET task status -> failed + return {"id": "task-9", "status": "failed", "error": "no speech detected"} + + monkeypatch.setattr(tool, "_request", fake_request) + + result = tool.execute( + { + "action": "caption", + "input_path": str(src), + "template_id": _TEMPLATES[0]["id"], + "poll_interval_seconds": 0, + } + ) + assert result.success is False + assert "no speech detected" in result.error diff --git a/tools/subtitle/zapcap_captions.py b/tools/subtitle/zapcap_captions.py new file mode 100644 index 000000000..a2e08d39c --- /dev/null +++ b/tools/subtitle/zapcap_captions.py @@ -0,0 +1,559 @@ +"""ZapCap captioning provider tool. + +ZapCap (https://zapcap.ai) is a video-captioning API: you upload a +video, pick a styled caption *template* (Hormozi, Beast, Devin, ...), and +ZapCap transcribes the audio and renders animated, word-synced subtitles +burned into the video. This tool wraps the full REST flow as a single +OpenMontage subtitle provider. + +End-to-end flow (the default ``action="caption"``): + + 1. Upload the video -> POST /videos (local file, multipart) + or POST /videos/url (public URL) -> videoId + 2. Create a render task -> POST /videos/{id}/task (templateId, ...) -> taskId + 3. Poll until completed -> GET /videos/{id}/task/{taskId} -> downloadUrl + 4. Download the rendered MP4 -> stream downloadUrl to output_path + +Granular actions (``list_templates``, ``upload``, ``create_task``, +``get_task``, ``approve_transcript``, ``balance``) expose each step so the +agent can drive fan-out flows (caption one video into N templates/languages +while transcribing only once via ``transcript_task_id``). + +## Configuration + +- ``ZAPCAP_API_KEY`` (required) — auth, sent as the ``x-api-key`` header. + +This matches the env-var contract of the official ``zapcap-mcp`` server so a +single ``.env`` configures both. +""" + +from __future__ import annotations + +import os +import time +from pathlib import Path +from typing import Any + +from tools.base_tool import ( + BaseTool, + Determinism, + ExecutionMode, + ResourceProfile, + RetryPolicy, + ToolResult, + ToolRuntime, + ToolStability, + ToolStatus, + ToolTier, +) + +BASE_URL = "https://api.zapcap.ai" + +# Terminal task states returned by GET /videos/{id}/task/{taskId} +_TERMINAL = {"completed", "failed"} + + +class ZapCapCaptions(BaseTool): + name = "zapcap_captions" + version = "0.1.0" + tier = ToolTier.CORE + capability = "subtitle" + provider = "zapcap" + stability = ToolStability.BETA + execution_mode = ExecutionMode.ASYNC # backend is async; execute() blocks via polling + determinism = Determinism.STOCHASTIC # transcription is ML-based + runtime = ToolRuntime.API + + dependencies = ["env:ZAPCAP_API_KEY", "python:requests"] + install_instructions = ( + "Set ZAPCAP_API_KEY to your ZapCap API key " + "(get one at https://platform.zapcap.ai/dashboard/api-key; requires a " + "Pro subscription + API credits)." + ) + # No OpenMontage fallback: ZapCap's styled-caption rendering is unique. + # For local burn-in use remotion_caption_burn (Remotion) or subtitle_gen (SRT/VTT). + fallback_tools = ["remotion_caption_burn", "subtitle_gen"] + agent_skills = ["zapcap-captions"] + + capabilities = [ + "upload_video", + "list_templates", + "create_caption_task", + "auto_caption", + "translate_captions", + "approve_transcript", + "burn_styled_captions", + ] + supports = { + "styled_templates": True, + "word_level_highlight": True, + "translation": True, + "emoji_captions": True, + "byo_transcript": True, + "offline": False, + } + best_for = [ + "social-ready animated, word-synced captions (TikTok/Reels/Shorts)", + "viral caption styles (Hormozi, Beast, Devin, ...) out of the box", + "translating + captioning the same video into many languages (fan-out)", + ] + not_good_for = [ + "fully offline production (requires the ZapCap API)", + "videos longer than 30 minutes (API limit)", + "non-speech videos (captions need an audio track to transcribe)", + ] + + input_schema = { + "type": "object", + "properties": { + "action": { + "type": "string", + "enum": [ + "caption", + "list_templates", + "upload", + "create_task", + "get_task", + "approve_transcript", + "balance", + ], + "default": "caption", + "description": ( + "'caption' (default) runs the full flow: upload -> task -> " + "poll -> download. The others expose individual steps for " + "fan-out or manual control." + ), + }, + # --- source (one of) --- + "input_path": { + "type": "string", + "description": "Absolute/relative path to a local video file (mp4 or mov).", + }, + "video_url": { + "type": "string", + "description": "Publicly reachable video/mp4 or video/quicktime URL.", + }, + "video_id": { + "type": "string", + "description": "An already-uploaded ZapCap videoId (skip the upload step).", + }, + # --- template --- + "template_id": { + "type": "string", + "description": "ZapCap template UUID (from list_templates).", + }, + "template_name": { + "type": "string", + "description": ( + "Template name (case-insensitive, e.g. 'Hormozi 1'). Resolved " + "to an id via /templates if template_id is not given." + ), + }, + # --- task options --- + "language": { + "type": "string", + "description": "ISO code of the source audio (e.g. 'en'). Auto-detected if omitted.", + }, + "translate_to": { + "type": "string", + "description": "Target language code to translate captions into (e.g. 'es', 'fr').", + }, + "transcript_task_id": { + "type": "string", + "description": ( + "Reuse an existing task's transcript instead of re-transcribing " + "(key for multi-template / multi-language fan-out)." + ), + }, + "auto_approve": { + "type": "boolean", + "default": True, + "description": "Auto-approve the transcript so rendering starts immediately.", + }, + "render_options": { + "type": "object", + "description": ( + "Subtitle appearance overrides passed through to ZapCap: " + "{subsOptions, styleOptions, highlightOptions}. See the " + "zapcap-captions skill for the full schema." + ), + }, + "transcribe_settings": { + "type": "object", + "description": "Transcription-time options (e.g. b-roll insertion).", + }, + "export_settings": { + "type": "object", + "description": "Export FPS/quality/codec/dimensions overrides.", + }, + "dictionary": { + "type": "array", + "items": {"type": "string"}, + "description": "Domain words/brand names to bias transcription toward.", + }, + "ttl": { + "type": "string", + "enum": ["1d", "7d", "30d"], + "description": "Retention for ZapCap-stored artifacts. Omit to keep indefinitely.", + }, + # --- task id (for get_task / approve_transcript) --- + "task_id": { + "type": "string", + "description": "Existing taskId (for get_task / approve_transcript).", + }, + # --- output + polling --- + "output_path": { + "type": "string", + "description": "Where to save the rendered captioned MP4 (action='caption').", + }, + "timeout_seconds": { + "type": "integer", + "default": 600, + "description": "Max seconds to poll for completion.", + }, + "poll_interval_seconds": { + "type": "number", + "default": 3.0, + "description": "Initial poll interval; doubles to a 15s cap.", + }, + }, + } + + resource_profile = ResourceProfile( + cpu_cores=1, ram_mb=256, vram_mb=0, disk_mb=500, network_required=True + ) + retry_policy = RetryPolicy(max_retries=2, retryable_errors=["rate_limit", "timeout"]) + idempotency_key_fields = [ + "input_path", + "video_url", + "video_id", + "template_id", + "template_name", + "language", + "translate_to", + ] + side_effects = [ + "uploads the video to the ZapCap API", + "creates a render task (consumes ZapCap credits)", + "downloads the rendered video to output_path", + ] + user_visible_verification = [ + "Play the output video and confirm captions are burned in and synced to speech", + "Verify the chosen template style matches expectation", + ] + + # ------------------------------------------------------------------ # + # Config / status + # ------------------------------------------------------------------ # + + def _api_key(self) -> str | None: + return os.environ.get("ZAPCAP_API_KEY") + + def get_status(self) -> ToolStatus: + if not self._api_key(): + return ToolStatus.UNAVAILABLE + try: + __import__("requests") + except ImportError: + return ToolStatus.UNAVAILABLE + return ToolStatus.AVAILABLE + + def _headers(self) -> dict[str, str]: + return {"x-api-key": self._api_key() or "", "accept": "application/json"} + + # ------------------------------------------------------------------ # + # Low-level HTTP + # ------------------------------------------------------------------ # + + def _request( + self, + method: str, + path: str, + *, + json_body: dict | None = None, + params: dict | None = None, + files: dict | None = None, + timeout: int = 120, + ) -> Any: + import requests + + url = f"{BASE_URL}{path if path.startswith('/') else '/' + path}" + headers = self._headers() + # Strip None query params + clean_params = {k: v for k, v in (params or {}).items() if v is not None} + resp = requests.request( + method, + url, + headers=headers, + json=json_body if files is None else None, + params=clean_params or None, + files=files, + timeout=timeout, + ) + if resp.status_code < 200 or resp.status_code >= 300: + body = resp.text[:500] + try: + j = resp.json() + if isinstance(j, dict) and "message" in j: + body = j["message"] + except Exception: + pass + raise RuntimeError( + f"ZapCap {method} {path} -> {resp.status_code} {resp.reason}: {body}" + ) + if resp.status_code == 204 or not resp.text: + return None + try: + return resp.json() + except Exception: + return resp.text + + # ------------------------------------------------------------------ # + # API operations + # ------------------------------------------------------------------ # + + def _list_templates(self) -> list[dict]: + data = self._request("GET", "/templates") + return data if isinstance(data, list) else [] + + def _resolve_template_id( + self, template_id: str | None, template_name: str | None + ) -> str: + if template_id: + return template_id + if not template_name: + raise ValueError("Provide template_id or template_name.") + templates = self._list_templates() + want = template_name.strip().lower() + for t in templates: + if str(t.get("name", "")).strip().lower() == want: + return t["id"] + available = ", ".join(sorted(str(t.get("name")) for t in templates)) + raise ValueError( + f"Template named {template_name!r} not found. Available: {available}" + ) + + def _upload( + self, + *, + input_path: str | None, + video_url: str | None, + ttl: str | None, + ) -> dict: + params = {"ttl": ttl} if ttl else None + if video_url: + return self._request("POST", "/videos/url", json_body={"url": video_url}, params=params) + if input_path: + p = Path(input_path) + if not p.is_file(): + raise FileNotFoundError(f"Video file not found: {input_path}") + with open(p, "rb") as fh: + files = {"file": (p.name, fh, "video/mp4")} + return self._request("POST", "/videos", files=files, params=params, timeout=600) + raise ValueError("Provide input_path or video_url to upload.") + + def _create_task(self, video_id: str, inputs: dict[str, Any]) -> dict: + template_id = self._resolve_template_id( + inputs.get("template_id"), inputs.get("template_name") + ) + body: dict[str, Any] = {"templateId": template_id} + # autoApprove defaults to True for agentic flows + body["autoApprove"] = inputs.get("auto_approve", True) + for key, api_key in ( + ("language", "language"), + ("translate_to", "translateTo"), + ("transcript_task_id", "transcriptTaskId"), + ("render_options", "renderOptions"), + ("transcribe_settings", "transcribeSettings"), + ("export_settings", "exportSettings"), + ("dictionary", "dictionary"), + ): + if inputs.get(key) is not None: + body[api_key] = inputs[key] + params = {"ttl": inputs["ttl"]} if inputs.get("ttl") else None + data = self._request( + "POST", f"/videos/{video_id}/task", json_body=body, params=params + ) + return {"taskId": data["taskId"], "templateId": template_id} + + def _get_task(self, video_id: str, task_id: str) -> dict: + return self._request("GET", f"/videos/{video_id}/task/{task_id}") + + def _approve_transcript(self, video_id: str, task_id: str) -> Any: + return self._request( + "POST", f"/videos/{video_id}/task/{task_id}/approve-transcript" + ) + + def _poll( + self, + video_id: str, + task_id: str, + *, + wait_for: str = "completed", + timeout_seconds: int = 600, + poll_interval_seconds: float = 3.0, + ) -> dict: + deadline = time.time() + timeout_seconds + interval = poll_interval_seconds + last: dict = {} + while True: + last = self._get_task(video_id, task_id) + status = last.get("status") + if status == "failed": + raise RuntimeError( + f"ZapCap task failed: {last.get('error') or last}" + ) + if status == wait_for: + return last + if wait_for == "transcriptionCompleted" and status in _TERMINAL: + return last + if time.time() >= deadline: + raise TimeoutError( + f"ZapCap task {task_id} did not reach '{wait_for}' within " + f"{timeout_seconds}s (last status: {status})." + ) + time.sleep(min(interval, max(0.0, deadline - time.time()))) + interval = min(interval * 2, 15.0) + + def _download(self, url: str, output_path: str) -> str: + import requests + + out = Path(output_path) + out.parent.mkdir(parents=True, exist_ok=True) + with requests.get(url, stream=True, timeout=600) as r: + r.raise_for_status() + with open(out, "wb") as fh: + for chunk in r.iter_content(chunk_size=1 << 16): + if chunk: + fh.write(chunk) + return str(out) + + # ------------------------------------------------------------------ # + # Execute + # ------------------------------------------------------------------ # + + def execute(self, inputs: dict[str, Any]) -> ToolResult: + if not self._api_key(): + return ToolResult( + success=False, + error="ZAPCAP_API_KEY not set. " + self.install_instructions, + ) + + action = inputs.get("action", "caption") + start = time.time() + try: + if action == "list_templates": + templates = self._list_templates() + return ToolResult( + success=True, + data={ + "count": len(templates), + "templates": [ + { + "id": t.get("id"), + "name": t.get("name"), + "categories": t.get("categories", []), + } + for t in templates + ], + }, + duration_seconds=round(time.time() - start, 2), + ) + + if action == "balance": + data = self._request("GET", "/credits/balance") + return ToolResult(success=True, data={"balance": data}) + + if action == "upload": + up = self._upload( + input_path=inputs.get("input_path"), + video_url=inputs.get("video_url"), + ttl=inputs.get("ttl"), + ) + return ToolResult( + success=True, + data={"videoId": up.get("id"), "status": up.get("status")}, + duration_seconds=round(time.time() - start, 2), + ) + + if action == "get_task": + task = self._get_task(inputs["video_id"], inputs["task_id"]) + return ToolResult(success=True, data=task) + + if action == "approve_transcript": + self._approve_transcript(inputs["video_id"], inputs["task_id"]) + return ToolResult(success=True, data={"approved": True}) + + if action == "create_task": + video_id = inputs["video_id"] + created = self._create_task(video_id, inputs) + return ToolResult( + success=True, + data={"videoId": video_id, **created}, + duration_seconds=round(time.time() - start, 2), + ) + + # --- default: full caption flow --- + return self._caption_flow(inputs, start) + + except Exception as exc: + return ToolResult( + success=False, + error=f"ZapCap {action} failed: {exc}", + duration_seconds=round(time.time() - start, 2), + ) + + def _caption_flow(self, inputs: dict[str, Any], start: float) -> ToolResult: + # 1. Resolve / upload to get a videoId + video_id = inputs.get("video_id") + upload_status = "reused" + if not video_id: + up = self._upload( + input_path=inputs.get("input_path"), + video_url=inputs.get("video_url"), + ttl=inputs.get("ttl"), + ) + video_id = up.get("id") + upload_status = up.get("status", "uploaded") + + # 2. Create the render task + created = self._create_task(video_id, inputs) + task_id = created["taskId"] + + # 3. Poll until completed + task = self._poll( + video_id, + task_id, + wait_for="completed", + timeout_seconds=inputs.get("timeout_seconds", 600), + poll_interval_seconds=inputs.get("poll_interval_seconds", 3.0), + ) + download_url = task.get("downloadUrl") + if not download_url: + return ToolResult( + success=False, + error=f"Task completed but no downloadUrl present: {task}", + ) + + # 4. Download the rendered video + output_path = inputs.get("output_path") + if not output_path: + output_path = f"zapcap_captioned_{task_id[:8]}.mp4" + local_path = self._download(download_url, output_path) + + return ToolResult( + success=True, + data={ + "videoId": video_id, + "taskId": task_id, + "templateId": created["templateId"], + "uploadStatus": upload_status, + "status": task.get("status"), + "downloadUrl": download_url, + "transcriptUrl": task.get("transcript"), + "output": local_path, + }, + artifacts=[local_path], + model="zapcap", + duration_seconds=round(time.time() - start, 2), + )