Skip to content

feat: add MuAPI video generation tool #196

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
Anil-matcha wants to merge 1 commit into
base: main
Choose a base branch
from
Open

feat: add MuAPI video generation tool #196

Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
204 changes: 204 additions & 0 deletions backend/director/tools/muapi_video.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,204 @@
import os
import time
import requests
from typing import Optional


PARAMS_CONFIG = {
"text_to_video": {
"model_name": {
"type": "string",
"description": "The muapi.ai model to use for text-to-video generation",
"default": "veo3-fast",
"enum": [
"veo3",
"veo3-fast",
"kling-master",
"wan2.1",
"wan2.2",
"seedance-pro",
"seedance-pro-fast",
"runway",
"pixverse",
"hunyuan",
"minimax-hailuo-02-std",
"minimax-hailuo-02-pro",
],
},
"duration": {
"type": "integer",
"description": "Duration of the video in seconds",
"default": 5,
"minimum": 3,
"maximum": 60,
},
"aspect_ratio": {
"type": "string",
"description": "Aspect ratio of the generated video",
"default": "16:9",
"enum": ["16:9", "9:16", "1:1", "4:3", "3:4"],
},
},
"image_to_video": {
"model_name": {
"type": "string",
"description": "The muapi.ai model to use for image-to-video generation",
"default": "kling-master",
"enum": [
"kling-master",
"kling-v2.5-pro",
"wan2.1",
"wan2.2",
"seedance-pro",
"seedance-pro-fast",
"runway",
"pixverse",
"hunyuan",
"vidu",
],
},
"duration": {
"type": "integer",
"description": "Duration of the video in seconds",
"default": 5,
"minimum": 3,
"maximum": 60,
},
"aspect_ratio": {
"type": "string",
"description": "Aspect ratio of the generated video",
"default": "16:9",
"enum": ["16:9", "9:16", "1:1", "4:3", "3:4"],
},
},
}

# Endpoints are the model IDs themselves for muapi
T2V_ENDPOINTS = {
"veo3": "veo3",
"veo3-fast": "veo3-fast",
"kling-master": "kling",
"wan2.1": "wan2.1",
"wan2.2": "wan2.2",
"seedance-pro": "seedance-pro",
"seedance-pro-fast": "seedance-pro-fast",
"runway": "runway",
"pixverse": "pixverse",
"hunyuan": "hunyuan",
"minimax-hailuo-02-std": "minimax-hailuo-02-std",
"minimax-hailuo-02-pro": "minimax-hailuo-02-pro",
}

I2V_ENDPOINTS = {
"kling-master": "kling-i2v",
"kling-v2.5-pro": "kling-v2.5-pro-i2v",
"wan2.1": "wan2.1-i2v",
"wan2.2": "wan2.2-i2v",
"seedance-pro": "seedance-pro-i2v",
"seedance-pro-fast": "seedance-pro-fast-i2v",
"runway": "runway-i2v",
"pixverse": "pixverse-i2v",
"hunyuan": "hunyuan-i2v",
"vidu": "vidu-i2v",
}


class MuApiVideoGenerationTool:
"""Video generation tool using muapi.ai's 400+ model aggregator API."""

BASE_URL = "https://api.muapi.ai/api/v1"
POLL_INTERVAL = 5 # seconds

def __init__(self, api_key: str):
if not api_key:
raise Exception(
"MUAPI_API_KEY not found. Get one at https://muapi.ai/dashboard/api-keys"
)
self.api_key = api_key
self.session = requests.Session()
self.session.headers.update(
{"x-api-key": self.api_key, "Content-Type": "application/json"}
)

def _submit(self, endpoint: str, payload: dict) -> str:
"""Submit a generation request and return the request_id."""
resp = self.session.post(f"{self.BASE_URL}/{endpoint}", json=payload, timeout=30)
resp.raise_for_status()
return resp.json()["request_id"]
Comment on lines +123 to +127

@coderabbitai coderabbitai Bot Jun 17, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

⚠️ Potential issue | 🟡 Minor | ⚡ Quick win

Handle missing request_id in API response.

If the API response doesn't contain the expected request_id key (e.g., due to an API change or unexpected error format), a raw KeyError is raised without context. Adding explicit handling improves debuggability.

Proposed fix 
     def _submit(self, endpoint: str, payload: dict) -> str:
         """Submit a generation request and return the request_id."""
         resp = self.session.post(f"{self.BASE_URL}/{endpoint}", json=payload, timeout=30)
         resp.raise_for_status()
-        return resp.json()["request_id"]
+        data = resp.json()
+        if "request_id" not in data:
+            raise Exception(f"API response missing 'request_id': {data}")
+        return data["request_id"]
📝 Committable suggestion

‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.

Suggested change
def _submit(self, endpoint: str, payload: dict) -> str:
"""Submit a generation request and return the request_id."""
resp = self.session.post(f"{self.BASE_URL}/{endpoint}", json=payload, timeout=30)
resp.raise_for_status()
return resp.json()["request_id"]
def _submit(self, endpoint: str, payload: dict) -> str:
"""Submit a generation request and return the request_id."""
resp = self.session.post(f"{self.BASE_URL}/{endpoint}", json=payload, timeout=30)
resp.raise_for_status()
data = resp.json()
if "request_id" not in data:
raise Exception(f"API response missing 'request_id': {data}")
return data["request_id"]
🤖 Prompt for AI Agents 
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@backend/director/tools/muapi_video.py` around lines 123 - 127, The _submit
method directly accesses resp.json()["request_id"] which will raise a raw
KeyError if the key is missing from the API response. Wrap the
resp.json()["request_id"] access in a try-except block to catch KeyError, and
when caught, raise a more informative exception that includes the actual
response content to aid debugging. This will provide clear context about what
the API returned instead of the expected request_id.


def _poll(self, request_id: str, timeout: int = 600) -> str:
"""Poll until the generation completes and return the output URL."""
deadline = time.time() + timeout
while time.time() < deadline:
resp = self.session.get(
f"{self.BASE_URL}/predictions/{request_id}/result", timeout=15
)
resp.raise_for_status()
data = resp.json()
status = data.get("status", "pending")
if status == "completed":
outputs = data.get("outputs", [])
if not outputs:
raise Exception("Completed but no outputs returned")
return outputs[0]
if status in ("failed", "cancelled"):
raise Exception(f"Video generation {status}: {data.get('error', '')}")
time.sleep(self.POLL_INTERVAL)
raise Exception(f"Video generation timed out after {timeout}s")

def text_to_video(
self, prompt: str, save_at: str, duration: float, config: dict
) -> dict:
"""Generate a video from a text prompt using muapi.ai."""
model_name = config.get("model_name", "veo3-fast")
endpoint = T2V_ENDPOINTS.get(model_name, model_name)

payload = {
"prompt": prompt,
"duration": int(duration),
"aspect_ratio": config.get("aspect_ratio", "16:9"),
}

try:
request_id = self._submit(endpoint, payload)
video_url = self._poll(request_id)
video_data = requests.get(video_url, timeout=120)
video_data.raise_for_status()
with open(save_at, "wb") as f:
f.write(video_data.content)
except Exception as e:
raise Exception(f"Error generating video: {type(e).__name__}: {str(e)}")

return {"status": "success", "video_path": save_at}
Comment on lines +162 to +172

@coderabbitai coderabbitai Bot Jun 17, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🛠️ Refactor suggestion | 🟠 Major | ⚡ Quick win

Preserve exception chain with raise ... from e.

The re-raised exception loses the original traceback, making debugging harder. Using exception chaining preserves the full context. The same issue applies to image_to_video (line 202).

Proposed fix for both methods 
         except Exception as e:
-            raise Exception(f"Error generating video: {type(e).__name__}: {str(e)}")
+            raise Exception(f"Error generating video: {type(e).__name__}: {e}") from e

Apply to both line 170 and line 202.

🧰 Tools
🪛 Ruff (0.15.17)

[warning] 169-169: Do not catch blind exception: Exception

(BLE001)

[warning] 170-170: Within an except clause, raise exceptions with raise ... from err or raise ... from None to distinguish them from errors in exception handling

(B904)

[warning] 170-170: Use explicit conversion flag

Replace with conversion flag

(RUF010)

🤖 Prompt for AI Agents 
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@backend/director/tools/muapi_video.py` around lines 162 - 172, The exception
handling in the video generation code is losing the original traceback by
re-raising a new Exception without preserving the exception chain. Modify the
raise statement in the except block (where the message "Error generating video"
is constructed) to use the `from e` syntax to preserve the original exception
context. Apply the same fix to the `image_to_video` method (mentioned at line
202) where this issue also exists. This ensures the full traceback is retained
for better debugging.

Source: Linters/SAST tools


def image_to_video(
self,
image_url: str,
save_at: str,
duration: float,
config: dict,
prompt: Optional[str] = None,
) -> dict:
"""Generate a video from an image URL using muapi.ai."""
model_name = config.get("model_name", "kling-master")
endpoint = I2V_ENDPOINTS.get(model_name, f"{model_name}-i2v")

payload: dict = {
"image_url": image_url,
"duration": int(duration),
"aspect_ratio": config.get("aspect_ratio", "16:9"),
}
if prompt:
payload["prompt"] = prompt

try:
request_id = self._submit(endpoint, payload)
video_url = self._poll(request_id)
video_data = requests.get(video_url, timeout=120)
video_data.raise_for_status()
with open(save_at, "wb") as f:
f.write(video_data.content)
except Exception as e:
raise Exception(f"Error generating video: {type(e).__name__}: {str(e)}")

return {"status": "success", "video_path": save_at}