import os
import time
import requests
api_key = os.environ["MINIMAX_API_KEY"]
headers = {"Authorization": f"Bearer {api_key}"}
BASE_URL = "https://api.minimax.io"
MODEL = "MiniMax-H3"
# --- Step 1: Create a video generation task ---
# MiniMax-H3 uses a multimodal content[] structure: each element is distinguished by type
# (text / image_url / video_url / audio_url) and can be labeled with a role. Each function below
# corresponds to one mode (text-to-video, image-to-video, first-and-last-frame, reference-to-video),
# starts an asynchronous task, and returns a unique task_id.
def invoke_text_to_video() -> str:
"""(Mode 1) Text-to-video (t2va). For t2va, ratio is required and cannot be 'adaptive'."""
url = f"{BASE_URL}/v2/video_generation"
payload = {
"model": MODEL,
"content": [
# A type=text item is required and defines the video's content and motion.
{"type": "text", "text": "A tiktok dancer is dancing on a drone, doing flips and tricks."},
],
"duration": 5,
"resolution": "2K",
"ratio": "16:9",
}
response = requests.post(url, headers=headers, json=payload)
response.raise_for_status()
return response.json()["task_id"]
def invoke_image_to_video() -> str:
"""(Mode 2) Image-to-video (i2va) using a first-frame image and text."""
url = f"{BASE_URL}/v2/video_generation"
payload = {
"model": MODEL,
"content": [
{"type": "text", "text": "Contemporary dance, the people in the picture are performing contemporary dance."},
# role=first_frame specifies the opening frame; for image-to-video the aspect ratio is
# determined by the input image and ratio is always 'adaptive'.
{"type": "image_url", "image_url": {"url": "https://filecdn.minimax.chat/public/85c96368-6ead-4eae-af9c-116be878eac3.png"}, "role": "first_frame"},
],
"duration": 5,
"resolution": "2K",
}
response = requests.post(url, headers=headers, json=payload)
response.raise_for_status()
return response.json()["task_id"]
def invoke_start_end_to_video() -> str:
"""(Mode 3) First-frame + last-frame image + text."""
url = f"{BASE_URL}/v2/video_generation"
payload = {
"model": MODEL,
"content": [
{"type": "text", "text": "A little girl grows up."},
# role=first_frame specifies the opening frame
{"type": "image_url", "image_url": {"url": "https://filecdn.minimax.chat/public/fe9d04da-f60e-444d-a2e0-18ae743add33.jpeg"}, "role": "first_frame"},
# role=last_frame specifies the ending frame
{"type": "image_url", "image_url": {"url": "https://filecdn.minimax.chat/public/97b7cd08-764e-4b8b-a7bf-87a0bd898575.jpeg"}, "role": "last_frame"},
],
"duration": 5,
"resolution": "2K",
}
response = requests.post(url, headers=headers, json=payload)
response.raise_for_status()
return response.json()["task_id"]
def invoke_reference_to_video() -> str:
"""(Mode 4) Reference-to-video (r2va): combine reference images / videos / audio."""
url = f"{BASE_URL}/v2/video_generation"
payload = {
"model": MODEL,
"content": [
{"type": "text", "text": "On an overcast day, in an ancient cobbled alleyway, the model walks and adjusts a vintage beret with a smile; natural lighting and cinematic colors."},
# role=reference_image provides a subject reference; you may also add
# role=reference_video / role=reference_audio as references.
{"type": "image_url", "image_url": {"url": "https://filecdn.minimax.chat/public/54be8fbe-5694-4422-9c95-99cf785eb90e.PNG"}, "role": "reference_image"},
],
"duration": 5,
"resolution": "2K",
}
response = requests.post(url, headers=headers, json=payload)
response.raise_for_status()
return response.json()["task_id"]
# --- Step 2: Poll task status ---
# Since video generation is time-consuming, the API works asynchronously.
# After submitting a task, poll its status using the task_id. On success the response directly
# returns the video download URL (content.url) — no file_id exchange is needed.
def query_task_status(task_id: str) -> str:
"""Poll task status by task_id and return the video download URL on success."""
url = f"{BASE_URL}/v2/query/video_generation/{task_id}"
while True:
# A recommended polling interval is 10 seconds to avoid unnecessary server load.
time.sleep(10)
response = requests.get(url, headers=headers)
response.raise_for_status()
task = response.json()["task"]
status = task["status"]
print(f"Current task status: {status}")
# On success, task.content.url is the video download URL.
if status == "succeeded":
return task["content"]["url"]
# Terminal failure states: failed / cancelled / expired.
if status in ("failed", "cancelled", "expired"):
raise Exception(f"Video generation did not succeed: status={status}, error={task.get('error')}")
# --- Step 3: Download and save the video file ---
# On success you get the download URL directly; download the content and save it locally.
def fetch_video(download_url: str):
"""Download the video and save it locally."""
with open("output.mp4", "wb") as f:
video_response = requests.get(download_url)
video_response.raise_for_status()
f.write(video_response.content)
print("Video successfully saved as output.mp4")
# --- Main process: end-to-end example ---
# Demonstrates the full workflow from task creation to video retrieval.
if __name__ == "__main__":
# Choose a task creation mode
task_id = invoke_text_to_video() # Mode 1: Text-to-Video
# task_id = invoke_image_to_video() # Mode 2: Image-to-Video
# task_id = invoke_start_end_to_video() # Mode 3: First-and-Last-Frame Video
# task_id = invoke_reference_to_video() # Mode 4: Reference-to-Video
print(f"Video generation task submitted, Task ID: {task_id}")
download_url = query_task_status(task_id)
print(f"Task succeeded, video URL: {download_url}")
fetch_video(download_url)
