generate_video
Create videos from images using AI models like SVD, AnimateDiff, or Kling. Specify image URL, model, and duration to generate video content through queued processing.
Instructions
Generate videos from images (uses queue API for long processing)
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| duration | No | Video duration in seconds | |
| image_url | Yes | Starting image URL (for image-to-video) | |
| model | No | Video generation model | svd |
Implementation Reference
- src/fal_mcp_server/server.py:219-266 (handler)Executes the generate_video tool by submitting an image-to-video job to the fal_client queue API and polling for completion using wait_for_queue_result.elif name == "generate_video": model_key = arguments.get("model", "svd") model_id = MODELS["video"][model_key] fal_args = {"image_url": arguments["image_url"]} if "duration" in arguments: fal_args["duration"] = arguments["duration"] # Submit to queue for processing handle = await fal_client.submit_async(model_id, arguments=fal_args) request_id = ( handle.request_id if hasattr(handle, "request_id") else str(handle) ) # Wait for completion with status updates response = f"⏳ Video generation queued (ID: {request_id[:8]}...)\n" response += "Processing (this may take 30-60 seconds)...\n" video_result: Optional[Dict[str, Any]] = await wait_for_queue_result( handle, timeout=180 ) if video_result is not None and "error" not in video_result: video_dict = video_result.get("video", {}) if isinstance(video_dict, dict): video_url = video_dict.get("url") else: video_url = video_result.get("url") if video_url: return [ TextContent( type="text", text=f"🎬 Video generated (via queue): {video_url}", ) ] else: error_msg = ( video_result.get("error", "Unknown error") if video_result else "Unknown error" ) return [ TextContent( type="text", text=f"❌ Video generation failed: {error_msg}", ) ]
- src/fal_mcp_server/server.py:127-152 (registration)Registers the generate_video tool with MCP server including input schema definition, available models from MODELS['video'], and parameters for image_url, model, duration.name="generate_video", description="Generate videos from images (uses queue API for long processing)", inputSchema={ "type": "object", "properties": { "image_url": { "type": "string", "description": "Starting image URL (for image-to-video)", }, "model": { "type": "string", "enum": list(MODELS["video"].keys()), "default": "svd", "description": "Video generation model", }, "duration": { "type": "integer", "default": 4, "minimum": 2, "maximum": 10, "description": "Video duration in seconds", }, }, "required": ["image_url"], }, ),
- src/fal_mcp_server/server.py:48-75 (helper)Helper function to poll the status of a queued fal_client job until completion, failure, or timeout. Used by generate_video handler.async def wait_for_queue_result( handle: Any, timeout: int = 300 ) -> Optional[Dict[str, Any]]: """Wait for a queued job to complete with timeout""" start_time = time.time() while True: # Check timeout if time.time() - start_time > timeout: return {"error": f"Timeout after {timeout} seconds"} # Check status using the handle status = await handle.status() if hasattr(status, "status"): status_str = status.status else: status_str = str(status) if "completed" in status_str.lower(): result = await handle.get() return cast(Dict[str, Any], result) elif "failed" in status_str.lower() or "error" in status_str.lower(): return {"error": f"Job failed: {status}"} # Wait before polling again await asyncio.sleep(2)