create_generation
Generate videos from text, images, or existing videos using customizable parameters like resolution, duration, and aspect ratio, powered by the MCP Luma Dream Machine server.
Instructions
Creates a new video generation from text, image, or existing video
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| aspect_ratio | No | ||
| callback_url | No | ||
| duration | No | ||
| keyframes | No | ||
| loop | No | ||
| model | No | ray-2 | |
| prompt | Yes | ||
| resolution | No |
Implementation Reference
- src/luma_ai_mcp_server/server.py:256-304 (handler)The handler function that implements the create_generation tool. It performs input validation, converts enum values if necessary, uses the CreateGenerationInput Pydantic model to structure the request, calls the Luma API to create a generation, and returns a formatted text response with the generation ID and state.async def create_generation(params: dict) -> str: """Create a new generation.""" if "prompt" not in params: raise ValueError("prompt parameter is required") if "model" in params: model = params["model"] if isinstance(model, str): if model not in [m.value for m in VideoModel]: raise ValueError(f"Invalid model: {model}") elif isinstance(model, VideoModel): params["model"] = model.value if "aspect_ratio" in params: aspect_ratio = params["aspect_ratio"] if isinstance(aspect_ratio, str): if aspect_ratio not in [a.value for a in AspectRatio]: raise ValueError(f"Invalid aspect ratio: {aspect_ratio}") elif isinstance(aspect_ratio, AspectRatio): params["aspect_ratio"] = aspect_ratio.value if "keyframes" in params: keyframes = params["keyframes"] if not isinstance(keyframes, dict): raise ValueError("keyframes must be an object") if not any(key in keyframes for key in ["frame0", "frame1"]): raise ValueError("keyframes must contain frame0 or frame1") input_data = CreateGenerationInput(**params) request_data = input_data.model_dump(exclude_none=True) response = await _make_luma_request("POST", "/generations", request_data) if input_data.keyframes: output = [ f"Created advanced generation with ID: {response['id']}", f"State: {response['state']}", ] if "frame0" in input_data.keyframes: output.append("starting from an image") if "frame1" in input_data.keyframes: output.append("ending with an image") else: output = [ f"Created text-to-video generation with ID: {response['id']}", f"State: {response['state']}", ] return "\n".join(output)
- Pydantic BaseModel defining the input schema for the create_generation tool, including required prompt and optional parameters like model, resolution, duration, aspect_ratio, loop, keyframes, and callback_url.class CreateGenerationInput(BaseModel): """ Input parameters for video generation. """ prompt: str model: VideoModel = VideoModel.RAY_2 resolution: Optional[Resolution] = None duration: Optional[Duration] = None aspect_ratio: Optional[AspectRatio] = None loop: Optional[bool] = None keyframes: Optional[dict] = None callback_url: Optional[str] = None
- src/luma_ai_mcp_server/server.py:503-506 (registration)Registration of the create_generation tool in the MCP server's list_tools() function, specifying name, description, and input schema.Tool( name=LumaTools.CREATE_GENERATION, description="Creates a new video generation from text, image, or existing video", inputSchema=CreateGenerationInput.model_json_schema(),
- src/luma_ai_mcp_server/server.py:559-561 (registration)Dispatch handler in the MCP server's call_tool() function that invokes the create_generation handler when the tool is called.case LumaTools.CREATE_GENERATION: result = await create_generation(arguments) return [TextContent(type="text", text=result)]
- Helper function used by create_generation to make authenticated HTTP requests to the Luma API.async def _make_luma_request(method: str, endpoint: str, data: Optional[dict] = None) -> dict: """Make a request to the Luma API.""" api_key = os.getenv("LUMA_API_KEY") if not api_key: raise ValueError("LUMA_API_KEY environment variable is not set") headers = { "Authorization": f"Bearer {api_key}", "Content-Type": "application/json", } async with httpx.AsyncClient(timeout=30.0) as client: try: response = await client.request( method, f"https://api.lumalabs.ai/dream-machine/v1{endpoint}", headers=headers, json=data if data else None, ) if response.status_code >= 400: error_msg = f"API request failed with status {response.status_code}" try: error_data = response.json() if "error" in error_data: error_msg = f"{error_msg}: {error_data['error']}" except Exception: pass raise ValueError(error_msg) return response.json() except httpx.NetworkError as e: logger.error(f"Network error occurred: {str(e)}") raise