diff --git a/tools/video/muapi_video.py b/tools/video/muapi_video.py new file mode 100644 index 00000000..a946c50a --- /dev/null +++ b/tools/video/muapi_video.py @@ -0,0 +1,255 @@ +"""MuAPI video generation via muapi.ai — 400+ model aggregator. + +Provides unified access to text-to-video and image-to-video generation +through muapi.ai's single API: Veo3, Kling, Wan, Seedance, Runway, and more. +""" + +from __future__ import annotations + +import os +import time +from pathlib import Path +from typing import Any + +import requests + +from tools.base_tool import ( + BaseTool, + Determinism, + ExecutionMode, + ResourceProfile, + RetryPolicy, + ToolResult, + ToolRuntime, + ToolStability, + ToolStatus, + ToolTier, +) + +BASE_URL = "https://api.muapi.ai/api/v1" + +T2V_MODELS = { + "veo3": "veo3", + "veo3-fast": "veo3-fast", + "kling-master": "kling", + "wan2.1": "wan2.1", + "wan2.2": "wan2.2", + "seedance-pro": "seedance-pro", + "seedance-pro-fast": "seedance-pro-fast", + "runway": "runway", + "pixverse": "pixverse", + "hunyuan": "hunyuan", + "minimax-hailuo-02-pro": "minimax-hailuo-02-pro", +} + +I2V_MODELS = { + "kling-master": "kling-i2v", + "kling-v2.5-pro": "kling-v2.5-pro-i2v", + "wan2.1": "wan2.1-i2v", + "wan2.2": "wan2.2-i2v", + "seedance-pro": "seedance-pro-i2v", + "runway": "runway-i2v", + "pixverse": "pixverse-i2v", + "hunyuan": "hunyuan-i2v", + "vidu": "vidu-i2v", +} + + +class MuApiVideo(BaseTool): + """Video generation via muapi.ai — a unified API for 400+ generative media models.""" + + name = "muapi_video" + version = "0.1.0" + tier = ToolTier.GENERATE + capability = "video_generation" + provider = "muapi" + stability = ToolStability.BETA + execution_mode = ExecutionMode.SYNC + determinism = Determinism.STOCHASTIC + runtime = ToolRuntime.API + + install_instructions = ( + "Set MUAPI_API_KEY to your muapi.ai API key.\n" + " Get one at https://muapi.ai/dashboard/api-keys" + ) + agent_skills = ["text-to-video", "image-to-video"] + + capabilities = ["text_to_video", "image_to_video"] + supports = { + "text_to_video": True, + "image_to_video": True, + "reference_image": True, + "native_audio": False, + "cinematic_quality": True, + "aspect_ratio": True, + "seed": False, + "offline": False, + } + best_for = [ + "access to 400+ video generation models through a single API", + "switching between Veo3, Kling, Wan, Seedance, Runway, and others without re-auth", + "cost-effective video generation with pay-per-use pricing", + ] + not_good_for = ["offline generation", "sub-second latency requirements"] + fallback_tools = ["kling_video", "runway_video", "seedance_video"] + + input_schema = { + "type": "object", + "required": ["prompt"], + "properties": { + "prompt": {"type": "string"}, + "operation": { + "type": "string", + "enum": ["text_to_video", "image_to_video"], + "default": "text_to_video", + }, + "model_variant": { + "type": "string", + "enum": sorted(T2V_MODELS), + "default": "veo3-fast", + "description": "Model to use. For image_to_video, valid values are: " + ", ".join(sorted(I2V_MODELS)), + }, + "duration": { + "type": "integer", + "default": 5, + "minimum": 3, + "maximum": 60, + "description": "Duration in seconds", + }, + "aspect_ratio": { + "type": "string", + "enum": ["16:9", "9:16", "1:1", "4:3", "3:4"], + "default": "16:9", + }, + "image_url": { + "type": "string", + "description": "Start frame image URL for image_to_video", + }, + "output_path": {"type": "string"}, + }, + } + + resource_profile = ResourceProfile( + cpu_cores=1, ram_mb=512, vram_mb=0, disk_mb=500, network_required=True + ) + retry_policy = RetryPolicy(max_retries=2, retryable_errors=["rate_limit", "timeout"]) + idempotency_key_fields = ["prompt", "model_variant", "operation", "duration"] + side_effects = ["writes video file to output_path", "calls muapi.ai API"] + user_visible_verification = [ + "Watch generated clip for motion coherence and visual quality" + ] + + def _get_api_key(self) -> str | None: + return os.environ.get("MUAPI_API_KEY") + + def get_status(self) -> ToolStatus: + if self._get_api_key(): + return ToolStatus.AVAILABLE + return ToolStatus.UNAVAILABLE + + def estimate_cost(self, inputs: dict[str, Any]) -> float: + # muapi.ai charges per second of video generated + duration = inputs.get("duration", 5) + return round(0.05 * duration, 2) + + def estimate_runtime(self, inputs: dict[str, Any]) -> float: + return 60.0 + + def _poll(self, api_key: str, request_id: str, timeout: int = 600) -> str: + headers = {"x-api-key": api_key} + deadline = time.time() + timeout + while time.time() < deadline: + resp = requests.get( + f"{BASE_URL}/predictions/{request_id}/result", + headers=headers, + timeout=15, + ) + resp.raise_for_status() + data = resp.json() + status = data.get("status", "pending") + if status == "completed": + outputs = data.get("outputs", []) + if not outputs: + raise ValueError("Generation completed but no outputs returned") + return outputs[0] + if status in ("failed", "cancelled"): + raise ValueError(f"Video generation {status}: {data.get('error', '')}") + time.sleep(5) + raise TimeoutError(f"Video generation timed out after {timeout}s") + + def execute(self, inputs: dict[str, Any]) -> ToolResult: + api_key = self._get_api_key() + if not api_key: + return ToolResult( + success=False, + error="MUAPI_API_KEY not set. " + self.install_instructions, + ) + + start = time.time() + operation = inputs.get("operation", "text_to_video") + model_variant = inputs.get("model_variant", "veo3-fast") + + if operation == "text_to_video": + endpoint = T2V_MODELS.get(model_variant, model_variant) + else: + endpoint = I2V_MODELS.get(model_variant, f"{model_variant}-i2v") + + payload: dict[str, Any] = { + "prompt": inputs["prompt"], + "duration": inputs.get("duration", 5), + "aspect_ratio": inputs.get("aspect_ratio", "16:9"), + } + + if operation == "image_to_video" and inputs.get("image_url"): + payload["image_url"] = inputs["image_url"] + + headers = {"x-api-key": api_key, "Content-Type": "application/json"} + + try: + submit_resp = requests.post( + f"{BASE_URL}/{endpoint}", + headers=headers, + json=payload, + timeout=30, + ) + submit_resp.raise_for_status() + request_id = submit_resp.json()["request_id"] + + video_url = self._poll(api_key, request_id) + + video_resp = requests.get(video_url, timeout=120) + video_resp.raise_for_status() + + output_path = Path(inputs.get("output_path", "muapi_output.mp4")) + output_path.parent.mkdir(parents=True, exist_ok=True) + output_path.write_bytes(video_resp.content) + + except Exception as e: + return ToolResult( + success=False, + error=f"MuAPI video generation failed: {e}", + ) + + from tools.video._shared import probe_output + + probed = probe_output(output_path) + return ToolResult( + success=True, + data={ + "provider": "muapi", + "model": model_variant, + "endpoint": endpoint, + "prompt": inputs["prompt"], + "operation": operation, + "aspect_ratio": inputs.get("aspect_ratio", "16:9"), + "duration": inputs.get("duration", 5), + "output": str(output_path), + "output_path": str(output_path), + "format": "mp4", + **probed, + }, + artifacts=[str(output_path)], + cost_usd=self.estimate_cost(inputs), + duration_seconds=round(time.time() - start, 2), + model=model_variant, + )