"""Google Veo 3.1 video generation via fal.ai API. Supports text-to-video, image-to-video, reference-to-video, and first/last-frame interpolation so agents can preserve visual consistency instead of relying only on raw text prompts. """ from __future__ import annotations import os import mimetypes import base64 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, ) class VeoVideo(BaseTool): name = "veo_video" version = "0.1.0" tier = ToolTier.GENERATE capability = "video_generation" provider = "veo" stability = ToolStability.EXPERIMENTAL execution_mode = ExecutionMode.SYNC determinism = Determinism.STOCHASTIC runtime = ToolRuntime.API dependencies = [] install_instructions = ( "Set FAL_KEY or FAL_AI_API_KEY to your fal.ai API key.\n" " Get one at https://fal.ai/dashboard/keys" ) agent_skills = ["ai-video-gen"] capabilities = ["text_to_video", "image_to_video", "reference_to_video", "first_last_frame_to_video"] supports = { "text_to_video": True, "image_to_video": True, "reference_to_video": True, "first_last_frame_to_video": True, "native_audio": True, "dialogue_generation": True, "ambient_sound": True, } best_for = [ "videos with synchronized dialogue and audio", "cutting-edge quality from Google DeepMind", "ambient sound and music generation built in", ] not_good_for = ["budget projects", "offline generation", "quick iteration"] fallback_tools = ["kling_video", "minimax_video", "wan_video"] input_schema = { "type": "object", "required": ["prompt"], "properties": { "prompt": {"type": "string"}, "operation": { "type": "string", "enum": ["text_to_video", "image_to_video", "reference_to_video", "first_last_frame_to_video"], "default": "text_to_video", }, "model_variant": { "type": "string", "enum": ["veo3", "veo3/fast", "veo3.1", "veo3.1/fast"], "default": "veo3.1", }, "duration": { "type": "string", "enum": ["4s", "6s", "8s"], "default": "8s", "description": "Duration in seconds", }, "aspect_ratio": { "type": "string", "enum": ["16:9", "9:16"], "default": "16:9", }, "generate_audio": { "type": "boolean", "default": True, "description": "Whether to generate synchronized audio", }, "resolution": { "type": "string", "enum": ["720p", "1080p", "4k"], "default": "1080p", }, "negative_prompt": {"type": "string"}, "seed": {"type": "integer"}, "auto_fix": {"type": "boolean", "default": True}, "safety_tolerance": { "type": "string", "enum": ["1", "2", "3", "4", "5", "6"], "default": "4", }, "image_url": {"type": "string", "description": "Reference image URL for image_to_video"}, "image_path": {"type": "string", "description": "Local reference image path for image_to_video"}, "reference_image_urls": { "type": "array", "items": {"type": "string"}, "description": "Reference image URLs for reference_to_video", }, "reference_image_paths": { "type": "array", "items": {"type": "string"}, "description": "Local reference image paths for reference_to_video", }, "first_frame_url": {"type": "string"}, "first_frame_path": {"type": "string"}, "last_frame_url": {"type": "string"}, "last_frame_path": {"type": "string"}, "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 fal.ai API"] user_visible_verification = [ "Watch generated clip for visual quality and motion", "Listen for audio synchronization and quality", ] def _get_api_key(self) -> str | None: return os.environ.get("FAL_KEY") or os.environ.get("FAL_AI_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: variant = inputs.get("model_variant", "veo3.1") duration_text = str(inputs.get("duration", "8s")).replace("s", "") duration = int(duration_text) resolution = inputs.get("resolution", "1080p") generate_audio = bool(inputs.get("generate_audio", True)) if "fast" in variant: base_per_second = 0.10 audio_per_second = 0.20 else: if resolution == "4k": base_per_second = 0.40 audio_per_second = 0.60 else: base_per_second = 0.20 audio_per_second = 0.40 return (audio_per_second if generate_audio else base_per_second) * duration def estimate_runtime(self, inputs: dict[str, Any]) -> float: variant = inputs.get("model_variant", "veo3.1") if "fast" in variant: return 45.0 return 120.0 @staticmethod def _file_to_data_uri(path_str: str) -> str: path = Path(path_str) if not path.exists(): raise FileNotFoundError(f"Input file not found: {path}") mime_type, _ = mimetypes.guess_type(path.name) if not mime_type: mime_type = "application/octet-stream" encoded = base64.b64encode(path.read_bytes()).decode("ascii") return f"data:{mime_type};base64,{encoded}" def _normalize_file_input(self, url_value: str | None, path_value: str | None) -> str | None: if url_value: return url_value if path_value: return self._file_to_data_uri(path_value) return None def execute(self, inputs: dict[str, Any]) -> ToolResult: api_key = self._get_api_key() if not api_key: return ToolResult( success=False, error="FAL_KEY / FAL_AI_API_KEY not set. " + self.install_instructions, ) import requests start = time.time() operation = inputs.get("operation", "text_to_video") variant = inputs.get("model_variant", "veo3.1") duration = inputs.get("duration", "8s") # Current fal Veo 3.1 image-guided endpoints only accept 8-second clips. if variant == "veo3.1" and operation in {"reference_to_video", "first_last_frame_to_video"} and duration != "8s": return ToolResult( success=False, error=( f"{operation} with {variant} currently requires duration='8s' on fal.ai; " f"received duration='{duration}'" ), ) # Build fal.ai model path operation_map = { "text_to_video": variant, "image_to_video": f"{variant}/image-to-video", "reference_to_video": f"{variant}/reference-to-video", "first_last_frame_to_video": f"{variant}/first-last-frame-to-video", } model_path = operation_map[operation] payload: dict[str, Any] = {"prompt": inputs["prompt"]} if inputs.get("duration"): payload["duration"] = inputs["duration"] if inputs.get("aspect_ratio"): payload["aspect_ratio"] = inputs["aspect_ratio"] if inputs.get("resolution"): payload["resolution"] = inputs["resolution"] if inputs.get("generate_audio") is not None: payload["generate_audio"] = inputs["generate_audio"] if inputs.get("negative_prompt"): payload["negative_prompt"] = inputs["negative_prompt"] if inputs.get("seed") is not None: payload["seed"] = inputs["seed"] if inputs.get("auto_fix") is not None: payload["auto_fix"] = inputs["auto_fix"] if inputs.get("safety_tolerance"): payload["safety_tolerance"] = inputs["safety_tolerance"] if operation == "image_to_video": image_value = self._normalize_file_input(inputs.get("image_url"), inputs.get("image_path")) if not image_value: return ToolResult( success=False, error="image_to_video requires image_url or image_path", ) payload["image_url"] = image_value if operation == "reference_to_video": image_urls = list(inputs.get("reference_image_urls") or []) image_paths = list(inputs.get("reference_image_paths") or []) normalized = list(image_urls) normalized.extend(self._file_to_data_uri(path) for path in image_paths) if not normalized: return ToolResult( success=False, error="reference_to_video requires reference_image_urls or reference_image_paths", ) payload["image_urls"] = normalized if operation == "first_last_frame_to_video": first_frame = self._normalize_file_input( inputs.get("first_frame_url"), inputs.get("first_frame_path") ) last_frame = self._normalize_file_input( inputs.get("last_frame_url"), inputs.get("last_frame_path") ) if not first_frame or not last_frame: return ToolResult( success=False, error="first_last_frame_to_video requires first_frame_url/path and last_frame_url/path", ) payload["first_frame_url"] = first_frame payload["last_frame_url"] = last_frame headers = { "Authorization": f"Key {api_key}", "Content-Type": "application/json", } try: # Submit to queue API (async) — sync endpoint times out for video gen submit_resp = requests.post( f"https://queue.fal.run/fal-ai/{model_path}", headers=headers, json=payload, timeout=30, ) submit_resp.raise_for_status() queue_data = submit_resp.json() status_url = queue_data["status_url"] response_url = queue_data["response_url"] # Poll until complete while True: time.sleep(5) status_resp = requests.get(status_url, headers=headers, timeout=15) status_resp.raise_for_status() status = status_resp.json().get("status", "UNKNOWN") if status == "COMPLETED": break if status in ("FAILED", "CANCELLED"): return ToolResult( success=False, error=f"Veo video generation {status.lower()}", ) # Fetch result result_resp = requests.get(response_url, headers=headers, timeout=30) if not result_resp.ok: detail = result_resp.text[:1000] return ToolResult( success=False, error=f"Veo video generation result fetch failed ({result_resp.status_code}): {detail}", ) data = result_resp.json() video_url = data["video"]["url"] video_response = requests.get(video_url, timeout=120) video_response.raise_for_status() output_path = Path(inputs.get("output_path", "veo_output.mp4")) output_path.parent.mkdir(parents=True, exist_ok=True) output_path.write_bytes(video_response.content) except Exception as e: return ToolResult(success=False, error=f"Veo video generation failed: {e}") return ToolResult( success=True, data={ "provider": "veo", "model": f"fal-ai/{model_path}", "prompt": inputs["prompt"], "output": str(output_path), "has_audio": inputs.get("generate_audio", True), "operation": operation, }, artifacts=[str(output_path)], cost_usd=self.estimate_cost(inputs), duration_seconds=round(time.time() - start, 2), model=f"fal-ai/{model_path}", )