comfyui-pipeline

ComfyUI Pipeline Skill

Interface ComfyUI as a programmable node in the Hermes asset pipeline. Run workflows, manage models, and feed outputs downstream to Blender or Godot.

ComfyUI API Base

Default: http://127.0.0.1:8188 Set COMFYUI_HOST env var to override.

Core Tools

1. Queue Prompt

import requests

def queue_comfyui_prompt(prompt_json: dict, workflow_name: str = "pipeline") -> str:
    """Queue a ComfyUI workflow. Returns prompt_id."""
    resp = requests.post(
        f"{COMFYUI_HOST}/api/prompt",
        json={"prompt": prompt_json, "workflow_name": workflow_name},
        timeout=30
    )
    resp.raise_for_status()
    return resp.json()["prompt_id"]

2. Monitor History

import requests, time

def wait_for_completion(prompt_id: str, timeout: int = 300) -> dict:
    """Poll ComfyUI history until prompt completes."""
    start = time.time()
    while time.time() - start < timeout:
        resp = requests.get(f"{COMFYUI_HOST}/api/history/{prompt_id}", timeout=10)
        history = resp.json()
        if prompt_id in history:
            status = history[prompt_id].get("status", {})
            if status.get("state") == "success":
                return history[prompt_id]
            elif status.get("state") == "failed":
                raise RuntimeError(f"Prompt failed: {status}")
        time.sleep(5)
    raise TimeoutError(f"Prompt {prompt_id} did not complete in {timeout}s")

3. Get Output Images

import os, shutil, requests
from pathlib import Path

def fetch_output_images(prompt_id: str, output_dir: Path) -> list[Path]:
    """Download all images from a completed ComfyUI run."""
    output_dir.mkdir(parents=True, exist_ok=True)
    resp = requests.get(f"{COMFYUI_HOST}/api/history/{prompt_id}", timeout=10)
    data = resp.json()[prompt_id]
    paths = []
    for node in data.get("outputs", {}).values():
        for item in node.get("images", []) + node.get("ui", {}).get("images", []):
            fname = item.get("filename")
            if fname:
                src = f"{COMFYUI_HOST}/view?filename={fname}&type=output"
                dst = output_dir / fname
                r = requests.get(src, timeout=60)
                dst.write_bytes(r.content)
                paths.append(dst)
    return paths