Adding Video Generation MCP tool, default model is Wan2.2-T2V-A14B

app.py

@@ -3,7 +3,6 @@
 #   2) DuckDuckGo Search — compact JSONL search output (short keys to minimize tokens)
 #   3) Python Code Executor — run Python code and capture stdout/errors
 #   4) Kokoro TTS — synthesize speech from text using Kokoro-82M
-#   5) Serverless Image Gen — Generate image on Hugging Face inference, default model is FLUX.1-Krea-dev.
 
 from __future__ import annotations
 
@@ -23,6 +22,7 @@ from urllib.parse import urljoin, urldefrag, urlparse
 from duckduckgo_search import DDGS
 from PIL import Image
 from huggingface_hub import InferenceClient
+import time
 
 # Optional imports for Kokoro TTS (loaded lazily)
 import numpy as np
@@ -631,7 +631,7 @@ CSS_STYLES = """
     }
     /* Default: add subtitle under titles */
     .gradio-container h1::after {
-    content: "Fetch Webpage | Search DuckDuckGo | Code Interpreter | Kokoro TTS | Image Generation";
+    content: "Fetch Webpage | Search DuckDuckGo | Code Interpreter | Kokoro TTS | Image Generation | Video Generation";
         display: block;
         font-size: 1rem;
         font-weight: 500;
@@ -792,15 +792,200 @@ image_generation_interface = gr.Interface(
     allow_flagging="never",
 )
 
+# ==========================
+# Video Generation (Serverless)
+# ==========================
+
+def _write_video_tmp(data_iter_or_bytes: object, suffix: str = ".mp4") -> str:
+    """Write video bytes or iterable of bytes to a temporary file and return its path."""
+    os.makedirs("outputs", exist_ok=True)
+    fname = f"outputs/video_{int(time.time())}_{random.randint(1000,9999)}{suffix}"
+    mode = "wb"
+    with open(fname, mode) as f:
+        # bytes-like
+        if isinstance(data_iter_or_bytes, (bytes, bytearray)):
+            f.write(data_iter_or_bytes)  # type: ignore[arg-type]
+        # file-like with read()
+        elif hasattr(data_iter_or_bytes, "read"):
+            f.write(data_iter_or_bytes.read())  # type: ignore[call-arg]
+        # response-like with content
+        elif hasattr(data_iter_or_bytes, "content"):
+            f.write(data_iter_or_bytes.content)  # type: ignore[attr-defined]
+        # iterable of chunks
+        elif hasattr(data_iter_or_bytes, "__iter__") and not isinstance(data_iter_or_bytes, (str, dict)):
+            for chunk in data_iter_or_bytes:  # type: ignore[assignment]
+                if chunk:
+                    f.write(chunk)
+        else:
+            raise gr.Error("Unsupported video data type returned by provider.")
+    return fname
+
+
+HF_VIDEO_TOKEN = os.getenv("HF_READ_TOKEN") or os.getenv("HF_TOKEN")
+
+
+def Generate_Video(  # <-- MCP tool #6 (Generate Video)
+    prompt: Annotated[str, "Text description of the video to generate (e.g., 'a red fox running through a snowy forest at sunrise')."],
+    model_id: Annotated[str, "Hugging Face model id in the form 'creator/model-name'. Defaults to Wan-AI/Wan2.2-T2V-A14B."] = "Wan-AI/Wan2.2-T2V-A14B",
+    negative_prompt: Annotated[str, "What should NOT appear in the video."] = "",
+    steps: Annotated[int, "Number of denoising steps (1–100). Higher can improve quality but is slower."] = 25,
+    cfg_scale: Annotated[float, "Guidance scale (1–20). Higher = follow the prompt more closely, lower = more creative."] = 3.5,
+    seed: Annotated[int, "Random seed for reproducibility. Use -1 for a random seed per call."] = -1,
+    width: Annotated[int, "Output width in pixels (multiples of 8 recommended)."] = 768,
+    height: Annotated[int, "Output height in pixels (multiples of 8 recommended)."] = 768,
+    fps: Annotated[int, "Frames per second of the output video (e.g., 24)."] = 24,
+    duration: Annotated[float, "Target duration in seconds (provider/model dependent, commonly 2–6s)."] = 4.0,
+) -> str:
+    """
+    Generate a short video from a text prompt using Hugging Face Inference Providers (Serverless Inference).
+
+    This tool follows the latest MCP guidance for Gradio-based MCP servers: clear type hints and
+    docstrings define the tool schema automatically. The returned file path will be converted to a file URL
+    for MCP clients.
+
+    Args:
+        prompt (str): Text description of the video to generate.
+    model_id (str): The Hugging Face model id (creator/model-name). Defaults to "Wan-AI/Wan2.2-T2V-A14B".
+        negative_prompt (str): What should NOT appear in the video.
+        steps (int): Number of denoising steps (1–100). Higher can improve quality but is slower.
+        cfg_scale (float): Guidance scale (1–20). Higher = follow the prompt more closely.
+        seed (int): Random seed. Use -1 to randomize on each call.
+        width (int): Output width in pixels.
+        height (int): Output height in pixels.
+        fps (int): Frames per second.
+        duration (float): Target duration in seconds.
+
+    Returns:
+        str: Path to an MP4 file on disk (Gradio will serve this file; MCP converts it to a file URL).
+
+    Error modes:
+        - Raises gr.Error with a user-friendly message on auth/model/load errors or unsupported parameters.
+    """
+    if not prompt or not prompt.strip():
+        raise gr.Error("Please provide a non-empty prompt.")
+
+    if not HF_VIDEO_TOKEN:
+        # Still attempt without a token (public models), but warn earlier if it fails.
+        pass
+
+    providers = ["auto", "replicate", "fal-ai"]
+    last_error: Exception | None = None
+
+    # Build a common parameters dict. Providers may ignore unsupported keys.
+    parameters = {
+        "negative_prompt": negative_prompt or None,
+        "num_inference_steps": steps,
+        "guidance_scale": cfg_scale,
+        "seed": seed if seed != -1 else random.randint(1, 1_000_000_000),
+        "width": width,
+        "height": height,
+        "fps": fps,
+        # Some providers/models expect num_frames instead of duration; we pass both-friendly value
+        # when supported; they may be ignored by the backend.
+        "duration": duration,
+    }
+
+    for provider in providers:
+        try:
+            client = InferenceClient(api_key=HF_VIDEO_TOKEN, provider=provider)
+            # Use the documented text_to_video API with correct parameters
+            if hasattr(client, "text_to_video"):
+                # Calculate num_frames from duration and fps if both provided
+                num_frames = int(duration * fps) if duration and fps else None
+                
+                # Build extra_body for provider-specific parameters
+                extra_body = {}
+                if width:
+                    extra_body["width"] = width
+                if height:
+                    extra_body["height"] = height
+                if fps:
+                    extra_body["fps"] = fps
+                if duration:
+                    extra_body["duration"] = duration
+                
+                result = client.text_to_video(
+                    prompt=prompt,
+                    model=model_id,
+                    guidance_scale=cfg_scale,
+                    negative_prompt=[negative_prompt] if negative_prompt else None,
+                    num_frames=num_frames,
+                    num_inference_steps=steps,
+                    seed=parameters["seed"],
+                    extra_body=extra_body if extra_body else None,
+                )
+            else:
+                # Generic POST fallback for older versions
+                result = client.post(
+                    model=model_id,
+                    json={
+                        "inputs": prompt,
+                        "parameters": {k: v for k, v in parameters.items() if v is not None},
+                    },
+                )
+
+            # Save output to an .mp4
+            path = _write_video_tmp(result, suffix=".mp4")
+            return path
+        except Exception as e:
+            last_error = e
+            continue
+
+    msg = str(last_error) if last_error else "Unknown error"
+    if "404" in msg:
+        raise gr.Error(f"Model not found or unavailable: {model_id}. Check the id and HF token access.")
+    if "503" in msg:
+        raise gr.Error("The model is warming up. Please try again shortly.")
+    if "401" in msg or "403" in msg:
+        raise gr.Error("Authentication failed or not permitted. Set HF_READ_TOKEN/HF_TOKEN with inference access.")
+    raise gr.Error(f"Video generation failed: {msg}")
+
+
+video_generation_interface = gr.Interface(
+    fn=Generate_Video,
+    inputs=[
+        gr.Textbox(label="Prompt", placeholder="Enter a prompt for the video", lines=2),
+    gr.Textbox(label="Model", value="Wan-AI/Wan2.2-T2V-A14B", placeholder="creator/model-name"),
+        gr.Textbox(label="Negative Prompt", value="", lines=2),
+        gr.Slider(minimum=1, maximum=100, value=25, step=1, label="Steps"),
+        gr.Slider(minimum=1.0, maximum=20.0, value=3.5, step=0.1, label="CFG Scale"),
+        gr.Slider(minimum=-1, maximum=1_000_000_000, value=-1, step=1, label="Seed (-1 = random)"),
+        gr.Slider(minimum=64, maximum=1920, value=768, step=8, label="Width"),
+        gr.Slider(minimum=64, maximum=1920, value=768, step=8, label="Height"),
+        gr.Slider(minimum=4, maximum=60, value=24, step=1, label="FPS"),
+        gr.Slider(minimum=1.0, maximum=10.0, value=4.0, step=0.5, label="Duration (s)"),
+    ],
+    outputs=gr.Video(label="Generated Video"),
+    title="Video Generation",
+    description=(
+    "<div style=\"text-align:center\">Generate short videos via Hugging Face Inference Providers. "
+    "Default model is Wan2.2-T2V-A14B.</div>"
+    ),
+    api_description=(
+        "Generate a short video from a text prompt using a Hugging Face model (Serverless Inference). "
+        "Parameters: prompt (str), model_id (str), negative_prompt (str), steps (int), cfg_scale (float), seed (int), "
+        "width/height (int), fps (int), duration (float). Returns a file path to an MP4 that MCP exposes as a file URL."
+    ),
+    allow_flagging="never",
+)
+
 # Build tabbed app including Image Generation
 demo = gr.TabbedInterface(
-    interface_list=[fetch_interface, concise_interface, code_interface, kokoro_interface, image_generation_interface],
+    interface_list=[
+        fetch_interface,
+        concise_interface,
+        code_interface,
+        kokoro_interface,
+        image_generation_interface,
+        video_generation_interface,
+    ],
     tab_names=[
         "Fetch Webpage",
         "DuckDuckGo Search",
         "Python Code Executor",
         "Kokoro TTS",
         "Image Generation",
+        "Video Generation",
     ],
     title="Tools MCP",
     theme="Nymbo/Nymbo_Theme",