Implementing a MiniMax-H3 Multimodal Video and Audio Generation Pipeline with ComfyUI APIs Comfy-Org released MiniMax-H3, a multimodal video and audio generation model, and a tutorial demonstrates implementing an end-to-end pipeline using ComfyUI as a headless inference backend. The pipeline supports text-to-video, first- and last-frame-conditioned generation, and reference-image-conditioned generation, with dynamic weight profile selection based on GPU memory (quality for 70+ GB, balanced for 38+ GB, squeeze for 20+ GB). The workflow uses HTTP and WebSocket APIs, constructs the execution graph in Python, and validates node schemas against the /object_info endpoint. In this tutorial, we implement an end-to-end MiniMax-H3 https://huggingface.co/Comfy-Org/MiniMax-H3 video generation workflow using ComfyUI as a headless inference backend. We configure the environment around GPU memory, disk capacity, model precision, resolution, duration, sampling strategy, and multiple generation modes, while dynamically selecting an appropriate weight profile based on the available hardware. We install and launch ComfyUI programmatically, download the required diffusion, text-encoder, video-VAE, and audio-VAE weights from Hugging Face, and communicate with the running server through its HTTP and WebSocket APIs. We also construct the ComfyUI execution graph directly in Python, validate node schemas against the live /object info endpoint, and support text-to-video, first- and last-frame-conditioned generation, and reference-image-conditioned generation. By combining automated model setup, schema-aware graph construction, joint video-audio decoding, progress monitoring, and output collection, we create a reproducible pipeline for experimenting with MiniMax-H3 without relying on the graphical ComfyUI interface. python import json, os, re, shutil, subprocess, sys, time, uuid, urllib.request, urllib.error from pathlib import Path CFG = { "MODE": "t2v", "PROMPT": "Realistic live-action cinematic look. A lone lighthouse keeper on a storm-lashed " "cliff at dusk, anamorphic lens, shallow depth of field, film grain, volumetric sea spray.\n" " 0s-2s Wide shot: waves detonate against black rock, the lighthouse beam sweeps the frame.\n" " 2s-4s Medium shot: the keeper braces against the wind, coat snapping, rain on his face.\n" " 4s-5s Close up: he squints into the dark and says \"She's holding.\"\n" "Camera: hard cuts between shots, slight handheld jitter, no dissolves.\n" "Audio: roaring surf and howling wind throughout, low cello drone underneath, " "a heavy wave impact on each cut, the line delivered clearly over the storm.\n" "No text, subtitles, logos or watermarks." , "ASPECT": 16, 9 , "MEGAPIXELS": 0.4, "SECONDS": 5.0, "SEED": 556589502035082, "STEPS": 20, "SAMPLER": "res multistep", "SCHEDULER": "simple", "FIRST FRAME": None, "LAST FRAME": None, "REF IMAGES": , "REF IMAGE SIZE": "match", "SIGMA SHIFT": None, "TURBO LORA": False, "TURBO STEPS": 8, "TURBO SAMPLER": "euler", "TURBO SCHEDULER": "beta", "COMFY DIR": "/content/ComfyUI", "OUT DIR": "/content/outputs", "MODELS ROOT": "/content/models", "PORT": 8188, "HF TOKEN": os.environ.get "HF TOKEN", "" , "SKIP INSTALL": False, } REPO = "Comfy-Org/MiniMax-H3" API = f"http://127.0.0.1:{CFG 'PORT' }" PROFILES = dict name="quality", min vram=70, unet fl="minimax h3 fl2va bf16.safetensors", unet ref="minimax h3 ref2va bf16.safetensors", te="qwen3vl 32b minimax h3 int8 convrot.safetensors", flags= "--normalvram" , dict name="balanced", min vram=38, unet fl="minimax h3 fl2va pruned int8 convrot.safetensors", unet ref="minimax h3 ref2va pruned int8 convrot.safetensors", te="qwen3vl 32b minimax h3 nvfp4 awq.safetensors", flags= "--normalvram", "--cache-none" , dict name="squeeze", min vram=20, unet fl="minimax h3 fl2va pruned fp8 scaled.safetensors", unet ref="minimax h3 ref2va pruned fp8 scaled.safetensors", te="qwen3vl 32b minimax h3 nvfp4 awq.safetensors", flags= "--lowvram", "--cache-none", "--disable-smart-memory" , VAE VIDEO = "minimax h3 video vae fp16.safetensors" VAE AUDIO = "minimax h3 audio vae fp32.safetensors" def sh cmd, cwd=None, check=True, quiet=False : """Run a shell command, streaming output.""" print f"$ {cmd}" p = subprocess.run cmd, shell=True, cwd=cwd, stdout=subprocess.DEVNULL if quiet else None, stderr=subprocess.STDOUT if quiet else None if check and p.returncode = 0: raise RuntimeError f"command failed {p.returncode} : {cmd}" def get json path, payload=None, timeout=30 : url = f"{API}{path}" data = json.dumps payload .encode if payload is not None else None req = urllib.request.Request url, data=data, headers={"Content-Type": "application/json"} with urllib.request.urlopen req, timeout=timeout as r: body = r.read return json.loads body if body else {} def align frames seconds, fps=24 : """H3 consumes frame counts on the 17k+5 grid. Snap upward.""" n = max 5, int round seconds fps while n % 17 = 5: n += 1 return n def h3 canvas aspect= 16, 9 , megapixels=0.98, multiple=32 : """Mirror of ComfyUI's ResolutionSelector + H3's 768 1344 area cap.""" ar = aspect 0 / aspect 1 total = megapixels 1e6 h = total / ar 0.5 w = ar h cap = 768 1344 if w h cap: s = cap / w h 0.5 w, h = w s, h s r = lambda v: max multiple, int round v / multiple multiple return r w , r h def preflight : try: import torch except ImportError: raise SystemExit "PyTorch missing — run this in a Colab GPU runtime." if not torch.cuda.is available : raise SystemExit "No CUDA device. Runtime Change runtime type GPU A100 ." name = torch.cuda.get device name 0 vram = torch.cuda.get device properties 0 .total memory / 1e9 free disk = shutil.disk usage "/content" .free / 1e9 bf16 = torch.cuda.is bf16 supported print f"GPU : {name} {vram:.1f} GB VRAM, bf16={bf16} " print f"Free disk : {free disk:.1f} GB" if not bf16: raise SystemExit "This GPU has no bf16 support T4/K80 . MiniMax-H3 will not run here.\n" "Switch to an A100/L4/H100 runtime." profile = next p for p in PROFILES if vram = p "min vram" , None if profile is None: raise SystemExit f"{vram:.0f} GB VRAM is below the ~20 GB floor for the smallest H3 build." if free disk < 45: print "WARNING: <45 GB free. Point MODELS ROOT at Drive or expect a disk-full error." print f"Profile : {profile 'name' } unet={profile 'unet fl' }, te={profile 'te' } " return profile We define the core MiniMax-H3 configuration, model profiles, generation parameters, and shared utility functions used throughout the workflow. We calculate valid frame counts and canvas dimensions while checking GPU capability, available VRAM, BF16 support, and disk space before inference begins. We also automatically select the most appropriate model profile so the pipeline matches the hardware available in our Colab runtime. python def install comfy : comfy = Path CFG "COMFY DIR" if CFG "SKIP INSTALL" and comfy.exists : print "Skipping install SKIP INSTALL=True ." return sh "pip install -q -U 'huggingface hub hf xet ' hf transfer websocket-client" if not comfy.exists : sh f"git clone --depth 1 https://github.com/comfyanonymous/ComfyUI {comfy}" sh f"pip install -q -r {comfy}/requirements.txt" ver = comfy / "comfyui version.py" if ver.exists : print "ComfyUI:", ver.read text .strip if not comfy / "comfy extras" / "nodes minimax h3.py" .exists : raise SystemExit "This ComfyUI checkout lacks native MiniMax-H3 nodes — update it." root = Path CFG "MODELS ROOT" for sub in "diffusion models", "text encoders", "vae", "loras" : root / sub .mkdir parents=True, exist ok=True comfy / "extra model paths.yaml" .write text "minimax h3:\n" f" base path: {root}\n" " diffusion models: diffusion models\n" " text encoders: text encoders\n" " vae: vae\n" " loras: loras\n" Path CFG "OUT DIR" .mkdir parents=True, exist ok=True def fetch repo id, filename, subdir : from huggingface hub import hf hub download os.environ "HF HUB ENABLE HF TRANSFER" = "1" dest = Path CFG "MODELS ROOT" / subdir target = dest / Path filename .name if target.exists and target.stat .st size 1 000 000: print f"cached {target.name} {target.stat .st size/1e9:.1f} GB " return target print f"pulling {filename} - {dest}" try: p = hf hub download repo id=repo id, filename=filename, local dir=str dest , token=CFG "HF TOKEN" or None except Exception as e: if "401" in str e or "403" in str e or "gated" in str e .lower : raise SystemExit f"Access denied for {repo id}. Accept the MiniMax-H3 community license on the " "model page, create a read token, then set CFG 'HF TOKEN' ." from e raise p = Path p if p = target: target.parent.mkdir parents=True, exist ok=True shutil.move str p , str target return target def download weights profile, mode : unet = profile "unet ref" if mode == "r2v" else profile "unet fl" fetch REPO, f"diffusion models/{unet}", "diffusion models" fetch REPO, f"text encoders/{profile 'te' }", "text encoders" fetch REPO, f"vae/{VAE VIDEO}", "vae" fetch REPO, f"vae/{VAE AUDIO}", "vae" lora = None if CFG "TURBO LORA" : from huggingface hub import HfApi lora repo = "drbaph/MiniMax-H3-Turbo-Lora-ComfyUI" files = f for f in HfApi .list repo files lora repo if f.endswith ".safetensors" and "pruned" in f if not files: files = f for f in HfApi .list repo files lora repo if f.endswith ".safetensors" if files: lora = fetch lora repo, sorted files -1 , "loras" .name print f"turbo LoRA: {lora}" return unet, profile "te" , lora We install and configure ComfyUI, prepare the external model directory structure, and enable MiniMax-H3 support inside the Colab environment. We download the required diffusion model, text encoder, video VAE, and audio VAE weights from Hugging Face while reusing cached files whenever possible. We also optionally retrieve the Turbo LoRA configuration, allowing us to trade some generation quality for faster inference when required. python class ComfyServer: def init self, flags : self.flags, self.proc, self.log = flags, None, Path "/content/comfyui.log" def start self : cmd = sys.executable, "main.py", "--listen", "127.0.0.1", "--port", str CFG "PORT" , "--disable-auto-launch", "--preview-method", "none", "--output-directory", CFG "OUT DIR" + self.flags print "$", " ".join cmd f = open self.log, "wb" self.proc = subprocess.Popen cmd, cwd=CFG "COMFY DIR" , stdout=f, stderr=subprocess.STDOUT deadline = time.time + 300 while time.time < deadline: if self.proc.poll is not None: print self.log.read text -4000: raise SystemExit "ComfyUI died during startup log above ." try: stats = get json "/system stats", timeout=3 dev = stats.get "devices", {} 0 print f"server up — {dev.get 'name','?' } " f"{dev.get 'vram total',0 /1e9:.1f} GB total, " f"{dev.get 'vram free',0 /1e9:.1f} GB free" return except Exception: time.sleep 2 raise SystemExit "Server did not become ready in 300s. Check /content/comfyui.log" def tail self, n=3000 : return self.log.read text -n: if self.log.exists else "" def free vram self : try: get json "/free", {"unload models": True, "free memory": True} except Exception: pass def stop self : if self.proc and self.proc.poll is None: self.proc.terminate try: self.proc.wait 30 except subprocess.TimeoutExpired: self.proc.kill class Schema: """Reads /object info so the graph is validated against the running node set instead of whatever the docs said last week.""" def init self : self.info = get json "/object info", timeout=120 def require self, classes : missing = c for c in classes if c not in self.info if missing: raise SystemExit f"Missing node classes: {missing}. Update ComfyUI to = 0.30.0." def inputs of self, cls : spec = self.info cls "input" return list spec.get "required", {} + list spec.get "optional", {} def check self, cls, payload : known = set self.inputs of cls unknown = k for k in payload if k not in known if unknown: print f" note: {cls} does not declare {unknown} — declared: {sorted known }" def autogrow self, cls, prefix, n : """Autogrow slots ref image 1, ref video 1, ... are dynamic; discover the real names if the server exposes them, otherwise fall back to 1-based.""" found = sorted k for k in self.inputs of cls if k.startswith prefix if len found = n: return found :n return f"{prefix}{i+1}" for i in range n We create a server-management layer that launches ComfyUI as a background subprocess and verifies that it becomes available through its API. We monitor server startup, inspect GPU memory statistics, free VRAM when necessary, and safely terminate the server after execution. We also build a schema-inspection utility that reads live ComfyUI node definitions so we can validate graph inputs and dynamically discover supported node slots. python class H3Graph: def init self, schema, unet, te, lora=None : self.s, self.g, self. id = schema, {}, 0 self.unet, self.te, self.lora = unet, te, lora def node self, cls, inputs : self.s.check cls, inputs self. id += 1 nid = str self. id self.g nid = {"class type": cls, "inputs": inputs} return nid def backbone self : model = self.node "UNETLoader", unet name=self.unet, weight dtype="default" if self.lora: model = self.node "LoraLoaderModelOnly", model= model, 0 , lora name=self.lora, strength model=1.0 if CFG "SIGMA SHIFT" : sv, sa = CFG "SIGMA SHIFT" model = self.node "MiniMaxH3SigmaShift", model= model, 0 , shift video=float sv , shift audio=float sa clip = self.node "CLIPLoader", clip name=self.te, type="minimax", device="default" vvae = self.node "VAELoader", vae name=VAE VIDEO avae = self.node "VAELoader", vae name=VAE AUDIO return model, clip, vvae, avae def tail self, model, cond, latent, vvae, avae : turbo = bool self.lora steps = CFG "TURBO STEPS" if turbo else CFG "STEPS" sampler name = CFG "TURBO SAMPLER" if turbo else CFG "SAMPLER" sched = CFG "TURBO SCHEDULER" if turbo else CFG "SCHEDULER" noise = self.node "RandomNoise", noise seed=int CFG "SEED" samp = self.node "KSamplerSelect", sampler name=sampler name sig = self.node "BasicScheduler", model= model, 0 , scheduler=sched, steps=steps, denoise=1.0 guider = self.node "BasicGuider", model= model, 0 , conditioning= cond 0 , cond 1 out = self.node "SamplerCustomAdvanced", noise= noise, 0 , guider= guider, 0 , sampler= samp, 0 , sigmas= sig, 0 , latent image= latent 0 , latent 1 frames = self.node "VAEDecode", samples= out, 0 , vae= vvae, 0 audio = self.node "VAEDecodeAudio", samples= out, 0 , vae= avae, 0 vid = self.node "CreateVideo", images= frames, 0 , audio= audio, 0 , fps=24 self.node "SaveVideo", video= vid, 0 , filename prefix="MiniMaxH3/h3", format="auto", codec="auto" print f" sampling: {steps} steps, {sampler name}/{sched}" return self.g def load image self, uploaded name : return self.node "LoadImage", image=uploaded name, upload="image" def t2v or flf2v self, w, h, length, first=None, last=None : self.s.require "MiniMaxH3ImageToVideo", "SamplerCustomAdvanced", "SaveVideo" model, clip, vvae, avae = self. backbone kw = {} if first: kw "first frame" = self. load image first , 0 if last: kw "last frame" = self. load image last , 0 n = self.node "MiniMaxH3ImageToVideo", clip= clip, 0 , vae= vvae, 0 , prompt=CFG "PROMPT" , width=w, height=h, length=length, kw return self. tail model, n, 0 , n, 1 , vvae, avae def r2v self, w, h, length, ref names : self.s.require "MiniMaxH3ReferenceToVideo" model, clip, vvae, avae = self. backbone slots = self.s.autogrow "MiniMaxH3ReferenceToVideo", "ref image ", len ref names refs = {slot: self. load image nm , 0 for slot, nm in zip slots, ref names } print f" reference slots: {list refs }" n = self.node "MiniMaxH3ReferenceToVideo", clip= clip, 0 , vae= vvae, 0 , audio vae= avae, 0 , prompt=CFG "PROMPT" , width=w, height=h, length=length, ref image size=CFG "REF IMAGE SIZE" , refs return self. tail model, n, 0 , n, 1 , vvae, avae We construct the MiniMax-H3 ComfyUI workflow graph entirely in Python using reusable node-building methods. We assemble the model backbone, conditioning pipeline, sampler, schedulers, joint latent decoding, video creation, and output-saving stages for both standard and Turbo configurations. We also support text-to-video, first- and last-frame-conditioned video, and reference-image-conditioned video generation through the same programmable graph architecture. python def upload image path : """Multipart POST to /upload/image; returns the name LoadImage expects.""" path = Path path if not path.exists : raise FileNotFoundError path boundary = uuid.uuid4 .hex body = f"--{boundary}\r\nContent-Disposition: form-data; name=\"image\"; " f"filename=\"{path.name}\"\r\nContent-Type: application/octet-stream\r\n\r\n" .encode + path.read bytes + f"\r\n--{boundary}\r\nContent-Disposition: form-data; name=\"overwrite\"\r\n\r\ntrue" f"\r\n--{boundary}--\r\n" .encode req = urllib.request.Request f"{API}/upload/image", data=body, headers={"Content-Type": f"multipart/form-data; boundary={boundary}"} with urllib.request.urlopen req, timeout=120 as r: info = json.loads r.read sub = info.get "subfolder" or "" print f" uploaded {path.name}" return f"{sub}/{info 'name' }" if sub else info "name" def run graph graph, server, timeout=7200 : """Submit, then follow the WebSocket for per-step progress.""" import websocket cid = uuid.uuid4 .hex Path "/content/last workflow api.json" .write text json.dumps graph, indent=2 try: res = get json "/prompt", {"prompt": graph, "client id": cid} except urllib.error.HTTPError as e: detail = e.read .decode :3000 raise SystemExit f"Graph rejected by ComfyUI:\n{detail}" pid = res "prompt id" print f"queued {pid} — first run loads ~37 GB of weights, be patient" ws = websocket.WebSocket ws.connect f"ws://127.0.0.1:{CFG 'PORT' }/ws?clientId={cid}", timeout=60 t0, last = time.time , "" try: while time.time - t0 < timeout: try: msg = ws.recv except Exception: time.sleep 1 continue if isinstance msg, bytes : continue d = json.loads msg t, data = d.get "type" , d.get "data", {} if t == "executing" and data.get "prompt id" == pid: if data.get "node" is None: print f"\ndone in {time.time -t0:.0f}s" break cls = graph.get data "node" , {} .get "class type", data "node" if cls = last: print f"\n - {cls}", end="", flush=True last = cls elif t == "progress": v, m = data.get "value", 0 , data.get "max", 1 print f"\r - {last} {v}/{m} ", end="", flush=True elif t == "execution error": print "\n--- execution error ---" print json.dumps data, indent=2 :4000 print server.tail raise SystemExit "Generation failed." finally: ws.close files = try: hist = get json f"/history/{pid}" for out in hist.get pid, {} .get "outputs", {} .values : for items in out.values : if isinstance items, list : for it in items: if isinstance it, dict and "filename" in it: p = Path CFG "OUT DIR" / it.get "subfolder" or "" / it "filename" if p.exists : files.append p except Exception: pass if not files: cands = p for p in Path CFG "OUT DIR" .rglob " " if p.suffix.lower in ".mp4", ".webm", ".mkv" and p.stat .st mtime t0 files = sorted cands, key=lambda p: p.stat .st mtime return files def main : profile = preflight install comfy mode = CFG "MODE" unet, te, lora = download weights profile, mode w, h = h3 canvas CFG "ASPECT" , CFG "MEGAPIXELS" length = align frames CFG "SECONDS" print f"\ncanvas {w}x{h}, {length} frames " f" {length/24:.2f}s @24fps, grid check {length % 17 == 5} " server = ComfyServer profile "flags" server.start try: schema = Schema builder = H3Graph schema, unet, te, lora if mode == "r2v": if not CFG "REF IMAGES" : raise SystemExit "MODE='r2v' needs CFG 'REF IMAGES' and