Fal.ai MCP Server

Instructions

Implementation Reference

• src/fal_mcp_server/handlers/video_handlers.py:209-378 (handler)

  Core handler function that processes tool arguments, resolves the AI model, prepares parameters for Fal.ai API, executes the video-to-video generation with timeout handling via queue_strategy, and returns the resulting video URL or detailed error message.
  async def handle_generate_video_from_video( arguments: Dict[str, Any], registry: ModelRegistry, queue_strategy: QueueStrategy, ) -> List[TextContent]: """Handle the generate_video_from_video tool for video-to-video transformations.""" model_input = arguments.get("model", "decart/lucy-edit/dev") try: model_id = await registry.resolve_model_id(model_input) except ValueError as e: return [ TextContent( type="text", text=f"❌ {e}. Use list_models to see available options.", ) ] # Build arguments based on model type fal_args: Dict[str, Any] = { "video_url": arguments["video_url"], "prompt": arguments["prompt"], } # Add optional parameters common to most models if "negative_prompt" in arguments: fal_args["negative_prompt"] = arguments["negative_prompt"] if "strength" in arguments: fal_args["strength"] = arguments["strength"] if "num_frames" in arguments: fal_args["num_frames"] = arguments["num_frames"] # Common video generation parameters if "duration" in arguments: fal_args["duration"] = arguments["duration"] if "aspect_ratio" in arguments: fal_args["aspect_ratio"] = arguments["aspect_ratio"] if "cfg_scale" in arguments: fal_args["cfg_scale"] = arguments["cfg_scale"] # Kling motion control specific parameters if "image_url" in arguments: fal_args["image_url"] = arguments["image_url"] if "character_orientation" in arguments: fal_args["character_orientation"] = arguments["character_orientation"] if "keep_original_sound" in arguments: fal_args["keep_original_sound"] = arguments["keep_original_sound"] # Kling Pro parameters (for transition videos and audio) if "tail_image_url" in arguments: fal_args["tail_image_url"] = arguments["tail_image_url"] if "generate_audio" in arguments: fal_args["generate_audio"] = arguments["generate_audio"] # Use queue strategy with extended timeout for video processing logger.info( "Starting video-to-video transformation with %s from %s", model_id, ( arguments["video_url"][:50] + "..." if len(arguments["video_url"]) > 50 else arguments["video_url"] ), ) try: # Video-to-video can take longer, use 300s timeout video_result = await asyncio.wait_for( queue_strategy.execute(model_id, fal_args, timeout=300), timeout=305, # Slightly longer than internal timeout ) except asyncio.TimeoutError: logger.error( "Video-to-video transformation timed out after 300s. Model: %s, Video: %s", model_id, ( arguments["video_url"][:50] + "..." if len(arguments["video_url"]) > 50 else arguments["video_url"] ), ) return [ TextContent( type="text", text=f"❌ Video transformation timed out after 300 seconds with {model_id}. Video processing can take several minutes for longer videos.", ) ] except Exception as e: logger.exception( "Video-to-video transformation failed. Model: %s, Video: %s", model_id, ( arguments["video_url"][:50] + "..." if len(arguments["video_url"]) > 50 else arguments["video_url"] ), ) return [ TextContent( type="text", text=f"❌ Video transformation failed: {e}", ) ] if video_result is None: logger.error( "Video-to-video transformation returned None. Model: %s, Video: %s", model_id, ( arguments["video_url"][:50] + "..." if len(arguments["video_url"]) > 50 else arguments["video_url"] ), ) return [ TextContent( type="text", text=f"❌ Video transformation failed or timed out with {model_id}", ) ] # Check for error in response if "error" in video_result: error_msg = video_result.get("error", "Unknown error") logger.error( "Video-to-video transformation failed for %s: %s", model_id, error_msg, ) return [ TextContent( type="text", text=f"❌ Video transformation failed: {error_msg}", ) ] # Extract video URL from result (handle different response formats) 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: source_preview = ( arguments["video_url"][:50] + "..." if len(arguments["video_url"]) > 50 else arguments["video_url"] ) return [ TextContent( type="text", text=f"🎬 Video transformed with {model_id}:\n\n**Source**: {source_preview}\n**Result**: {video_url}", ) ] logger.warning( "Video transformation completed but no video URL in response. Model: %s, Video: %s, Response keys: %s", model_id, ( arguments["video_url"][:50] + "..." if len(arguments["video_url"]) > 50 else arguments["video_url"] ), list(video_result.keys()) if video_result else "None", ) return [ TextContent( type="text", text="❌ Video transformation completed but no video URL was returned. Please try again.", ) ]
• src/fal_mcp_server/tools/video_tools.py:100-178 (schema)

  Defines the tool schema with name, detailed description, and comprehensive inputSchema specifying all parameters (required: video_url, prompt; optionals: model, duration, strength, Kling-specific params like image_url, etc.) for input validation.
  Tool( name="generate_video_from_video", description="Transform an existing video using AI. Supports restyling (Lucy models) and motion transfer (Kling motion control). Use upload_file first if you have a local video.", inputSchema={ "type": "object", "properties": { "video_url": { "type": "string", "description": "URL of the source video to transform (use upload_file for local videos)", }, "prompt": { "type": "string", "description": "Text description of the transformation (e.g., 'transform into anime style', 'a woman dancing gracefully')", }, "model": { "type": "string", "default": "decart/lucy-edit/dev", "description": "Video-to-video model. Options: decart/lucy-edit/dev (restyle), decart/lucy-edit/pro, decart/lucy-restyle, fal-ai/kling-video/v2.6/standard/motion-control (motion transfer), fal-ai/kling-video/v2.6/pro/motion-control", }, "duration": { "type": "integer", "enum": [5, 10], "default": 5, "description": "Duration of generated video in seconds", }, "aspect_ratio": { "type": "string", "enum": ["16:9", "9:16", "1:1"], "default": "16:9", "description": "Aspect ratio of the generated video", }, "cfg_scale": { "type": "number", "default": 0.5, "description": "Classifier Free Guidance scale - how closely to follow the prompt (0.0-1.0)", }, "image_url": { "type": "string", "description": "[Kling motion control] Reference image URL. The character in this image will be animated using motion from video_url.", }, "character_orientation": { "type": "string", "enum": ["image", "video"], "default": "video", "description": "[Kling motion control] 'video': orientation matches reference video (max 30s). 'image': orientation matches reference image (max 10s).", }, "keep_original_sound": { "type": "boolean", "default": True, "description": "[Kling motion control] Whether to keep original sound from reference video.", }, "tail_image_url": { "type": "string", "description": "[Kling Pro] URL of image for the end of the video (for transitions).", }, "generate_audio": { "type": "boolean", "default": True, "description": "[Kling v2.6 Pro] Generate native audio for video (supports Chinese/English).", }, "negative_prompt": { "type": "string", "description": "What to avoid in the output (default: 'blur, distort, and low quality')", }, "strength": { "type": "number", "minimum": 0, "maximum": 1, "default": 0.75, "description": "[Lucy models] How much to transform (0=keep original, 1=full transformation)", }, "num_frames": { "type": "integer", "description": "[Lucy models] Number of frames to process", }, }, "required": ["video_url", "prompt"], }, ),
• src/fal_mcp_server/server.py:61-85 (registration)

  Registers the tool by mapping the name 'generate_video_from_video' to its handler in the TOOL_HANDLERS dictionary, which is used in the @server.call_tool() method to route tool calls to the correct implementation.
  TOOL_HANDLERS = { # Utility tools (no queue needed) "list_models": handle_list_models, "recommend_model": handle_recommend_model, "get_pricing": handle_get_pricing, "get_usage": handle_get_usage, "upload_file": handle_upload_file, # Image generation tools "generate_image": handle_generate_image, "generate_image_structured": handle_generate_image_structured, "generate_image_from_image": handle_generate_image_from_image, # Image editing tools "remove_background": handle_remove_background, "upscale_image": handle_upscale_image, "edit_image": handle_edit_image, "inpaint_image": handle_inpaint_image, "resize_image": handle_resize_image, "compose_images": handle_compose_images, # Video tools "generate_video": handle_generate_video, "generate_video_from_image": handle_generate_video_from_image, "generate_video_from_video": handle_generate_video_from_video, # Audio tools "generate_music": handle_generate_music, }
• src/fal_mcp_server/handlers/video_handlers.py:33-209 (registration)

  No, wrong. Wait, this is the handler file itself.
  fal_args: Dict[str, Any] = { "prompt": arguments["prompt"], } # image_url is optional - only needed for image-to-video models if "image_url" in arguments: fal_args["image_url"] = arguments["image_url"] if "duration" in arguments: fal_args["duration"] = arguments["duration"] if "aspect_ratio" in arguments: fal_args["aspect_ratio"] = arguments["aspect_ratio"] if "negative_prompt" in arguments: fal_args["negative_prompt"] = arguments["negative_prompt"] if "cfg_scale" in arguments: fal_args["cfg_scale"] = arguments["cfg_scale"] # Use queue strategy with timeout protection for long-running video generation logger.info("Starting video generation with %s", model_id) try: video_result = await asyncio.wait_for( queue_strategy.execute(model_id, fal_args, timeout=180), timeout=185, # Slightly longer than internal timeout ) except asyncio.TimeoutError: return [ TextContent( type="text", text=f"❌ Video generation timed out after 180 seconds with {model_id}", ) ] if video_result is None: return [ TextContent( type="text", text=f"❌ Video generation failed or timed out with {model_id}", ) ] # Check for error in response if "error" in video_result: error_msg = video_result.get("error", "Unknown error") return [ TextContent( type="text", text=f"❌ Video generation failed: {error_msg}", ) ] # Extract video URL from 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 with {model_id}: {video_url}", ) ] return [ TextContent( type="text", text="❌ Video generation completed but no video URL was returned. Please try again.", ) ] async def handle_generate_video_from_image( arguments: Dict[str, Any], registry: ModelRegistry, queue_strategy: QueueStrategy, ) -> List[TextContent]: """Handle the generate_video_from_image tool.""" model_input = arguments.get("model", "fal-ai/wan-i2v") try: model_id = await registry.resolve_model_id(model_input) except ValueError as e: return [ TextContent( type="text", text=f"❌ {e}. Use list_models to see available options.", ) ] # Both image_url and prompt are required for this tool fal_args: Dict[str, Any] = { "image_url": arguments["image_url"], "prompt": arguments["prompt"], } if "duration" in arguments: fal_args["duration"] = arguments["duration"] if "aspect_ratio" in arguments: fal_args["aspect_ratio"] = arguments["aspect_ratio"] if "negative_prompt" in arguments: fal_args["negative_prompt"] = arguments["negative_prompt"] if "cfg_scale" in arguments: fal_args["cfg_scale"] = arguments["cfg_scale"] # Use queue strategy with timeout protection logger.info( "Starting image-to-video generation with %s from %s", model_id, ( arguments["image_url"][:50] + "..." if len(arguments["image_url"]) > 50 else arguments["image_url"] ), ) try: video_result = await asyncio.wait_for( queue_strategy.execute(model_id, fal_args, timeout=180), timeout=185, # Slightly longer than internal timeout ) except asyncio.TimeoutError: logger.error( "Image-to-video generation timed out after 180s. Model: %s, Image: %s", model_id, ( arguments["image_url"][:50] + "..." if len(arguments["image_url"]) > 50 else arguments["image_url"] ), ) return [ TextContent( type="text", text=f"❌ Video generation timed out after 180 seconds with {model_id}", ) ] if video_result is None: return [ TextContent( type="text", text=f"❌ Video generation failed or timed out with {model_id}", ) ] # Check for error in response if "error" in video_result: error_msg = video_result.get("error", "Unknown error") return [ TextContent( type="text", text=f"❌ Video generation failed: {error_msg}", ) ] # Extract video URL from 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 with {model_id}: {video_url}", ) ] return [ TextContent( type="text", text="❌ Video generation completed but no video URL was returned. Please try again.", ) ] async def handle_generate_video_from_video(