Fal.ai MCP Server

Instructions

Implementation Reference

• src/fal_mcp_server/server.py:267-319 (handler)

  Handler implementation for 'generate_music' tool: selects model from MODELS['audio'], submits async queue job to fal_client with prompt and duration, polls status with wait_for_queue_result, extracts and returns audio URL or error message.
  elif name == "generate_music": model_key = arguments.get("model", "musicgen") model_id = MODELS["audio"][model_key] # Submit to queue for longer music generation handle = await fal_client.submit_async( model_id, arguments={ "prompt": arguments["prompt"], "duration_seconds": arguments.get("duration_seconds", 30), }, ) request_id = ( handle.request_id if hasattr(handle, "request_id") else str(handle) ) duration = arguments.get("duration_seconds", 30) response = f"⏳ Music generation queued (ID: {request_id[:8]}...)\n" response += ( f"Generating {duration}s of music (this may take 20-40 seconds)...\n" ) music_result: Optional[Dict[str, Any]] = await wait_for_queue_result( handle, timeout=120 ) if music_result is not None and "error" not in music_result: audio_dict = music_result.get("audio", {}) if isinstance(audio_dict, dict): audio_url = audio_dict.get("url") else: audio_url = music_result.get("audio_url") if audio_url: return [ TextContent( type="text", text=f"🎵 Generated {duration}s of music (via queue): {audio_url}", ) ] else: error_msg = ( music_result.get("error", "Unknown error") if music_result else "Unknown error" ) return [ TextContent( type="text", text=f"❌ Music generation failed: {error_msg}", ) ]
• src/fal_mcp_server/server.py:153-179 (schema)

  Input schema for 'generate_music' tool defining required 'prompt' and optional 'duration_seconds' and 'model' parameters.
  Tool( name="generate_music", description="Generate music from text descriptions (uses queue API for long processing)", inputSchema={ "type": "object", "properties": { "prompt": { "type": "string", "description": "Description of the music (genre, mood, instruments)", }, "duration_seconds": { "type": "integer", "default": 30, "minimum": 5, "maximum": 300, "description": "Duration in seconds", }, "model": { "type": "string", "enum": ["musicgen", "musicgen_large"], "default": "musicgen", "description": "Music generation model", }, }, "required": ["prompt"], }, ),
• src/fal_mcp_server/server.py:77-180 (registration)

  Registration of 'generate_music' tool via @server.list_tools() decorator, including it in the list of available tools.
  @server.list_tools() async def list_tools() -> List[Tool]: """List all available Fal.ai tools""" return [ Tool( name="generate_image", description="Generate images from text prompts using various models (fast, uses async API)", inputSchema={ "type": "object", "properties": { "prompt": { "type": "string", "description": "Text description of the image to generate", }, "model": { "type": "string", "enum": list(MODELS["image"].keys()), "default": "flux_schnell", "description": "Model to use for generation", }, "negative_prompt": { "type": "string", "description": "What to avoid in the image", }, "image_size": { "type": "string", "enum": [ "square", "landscape_4_3", "landscape_16_9", "portrait_3_4", "portrait_9_16", ], "default": "landscape_16_9", }, "num_images": { "type": "integer", "default": 1, "minimum": 1, "maximum": 4, }, "seed": { "type": "integer", "description": "Seed for reproducible generation", }, }, "required": ["prompt"], }, ), Tool( 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"], }, ), Tool( name="generate_music", description="Generate music from text descriptions (uses queue API for long processing)", inputSchema={ "type": "object", "properties": { "prompt": { "type": "string", "description": "Description of the music (genre, mood, instruments)", }, "duration_seconds": { "type": "integer", "default": 30, "minimum": 5, "maximum": 300, "description": "Duration in seconds", }, "model": { "type": "string", "enum": ["musicgen", "musicgen_large"], "default": "musicgen", "description": "Music generation model", }, }, "required": ["prompt"], }, ), ]
• src/fal_mcp_server/server.py:48-75 (helper)

  Helper function used by generate_music handler to poll the fal_client queue job status until completion, failure, or timeout.
  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)
• src/fal_mcp_server/server.py:26-45 (helper)

  Global MODELS dictionary defining available models, including audio models ('musicgen', 'musicgen_large') used by generate_music.
  MODELS = { "image": { "flux_schnell": "fal-ai/flux/schnell", "flux_dev": "fal-ai/flux/dev", "flux_pro": "fal-ai/flux-pro", "sdxl": "fal-ai/fast-sdxl", "stable_diffusion": "fal-ai/stable-diffusion-v3-medium", }, "video": { "svd": "fal-ai/stable-video-diffusion", "animatediff": "fal-ai/fast-animatediff", "kling": "fal-ai/kling-video", }, "audio": { "musicgen": "fal-ai/musicgen-medium", "musicgen_large": "fal-ai/musicgen-large", "bark": "fal-ai/bark", "whisper": "fal-ai/whisper", }, }