ossrules.md

comfyui

Generate images, video, and audio via diffusion workflows.

  • comfyui
  • image-generation
  • stable-diffusion
  • flux
  • sd3
  • wan-video
  • hunyuan-video
  • creative
  • generative-ai
  • video-generation

Declared platforms: macos · linux · windows

Install
npx skills add 'https://github.com/NousResearch/hermes-agent/tree/main/optional-skills/creative/comfyui'
Download bundle ↓
main · 24fd22b ↗Scanned 2026-09-15

Contributors

GitHub-linked commit authors for this SKILL.md at the saved revision. Co-authors and history before file renames are not included.

File history ↗

scripts/ws_monitor.py

scripts/ws_monitor.pyBrowse 33 files
View on GitHub
2,476 tokens

10,512 bytes
Token encoding: o200k_base
Snapshot 24fd22b

← Back to SKILL.md
#!/usr/bin/env python3"""ws_monitor.py — Real-time ComfyUI WebSocket monitor. Connects to /ws and pretty-prints execution events: node start/finish, samplingprogress, cached nodes, errors. Optionally writes preview frames to disk. Useful for:  - Watching a long-running job in real time without parsing JSON yourself  - Saving in-progress preview frames for video / animation workflows  - Debugging "why is this hanging?" — see exactly which node is stuck Usage:    # Local — watch all jobs from this client_id    python3 ws_monitor.py     # Cloud — watch a specific prompt_id    python3 ws_monitor.py --host https://cloud.comfy.org \        --prompt-id abc-123-def     # Save preview frames to ./previews/    python3 ws_monitor.py --previews ./previews Requires: websocket-client (`pip install websocket-client`).Falls back to a clear error message when not installed.""" from __future__ import annotations import argparseimport jsonimport structimport sysfrom pathlib import Pathfrom urllib.parse import urlparse sys.path.insert(0, str(Path(__file__).resolve().parent))from _common import (  # noqa: E402    DEFAULT_LOCAL_HOST, ENV_API_KEY, log, new_client_id, resolve_api_key, is_cloud_host,)  # Binary frame types from ComfyUI WebSocket protocolBINARY_PREVIEW_IMAGE = 1BINARY_TEXT = 3BINARY_PREVIEW_IMAGE_WITH_METADATA = 4 # Image type codes inside PREVIEW_IMAGEIMAGE_TYPE_JPEG = 1IMAGE_TYPE_PNG = 2 # ANSI escape codes (works on most modern terminals)RESET = "\033[0m"DIM = "\033[2m"BOLD = "\033[1m"GREEN = "\033[32m"YELLOW = "\033[33m"RED = "\033[31m"CYAN = "\033[36m"  def fmt_color(s: str, color: str, *, color_on: bool = True) -> str:    return f"{color}{s}{RESET}" if color_on else s  def parse_binary_frame(data: bytes) -> dict | None:    if len(data) < 8:        return None    type_code = struct.unpack(">I", data[0:4])[0]    if type_code == BINARY_PREVIEW_IMAGE:        image_type = struct.unpack(">I", data[4:8])[0]        ext = "jpg" if image_type == IMAGE_TYPE_JPEG else "png" if image_type == IMAGE_TYPE_PNG else "bin"        return {            "kind": "preview",            "image_type": image_type,            "ext": ext,            "image_bytes": data[8:],        }    if type_code == BINARY_PREVIEW_IMAGE_WITH_METADATA:        if len(data) < 12:            return None        meta_len = struct.unpack(">I", data[4:8])[0]        meta_end = 8 + meta_len        if len(data) < meta_end:            return None        try:            meta = json.loads(data[8:meta_end].decode("utf-8"))        except Exception:            meta = {"raw": data[8:meta_end][:200].decode("utf-8", "replace")}        return {            "kind": "preview_with_metadata",            "metadata": meta,            "image_bytes": data[meta_end:],            "ext": "png",        }    if type_code == BINARY_TEXT:        if len(data) < 8:            return None        nid_len = struct.unpack(">I", data[4:8])[0]        nid_end = 8 + nid_len        if len(data) < nid_end:            return None        return {            "kind": "text",            "node_id": data[8:nid_end].decode("utf-8", "replace"),            "text": data[nid_end:].decode("utf-8", "replace"),        }    return {"kind": "unknown", "type_code": type_code, "size": len(data)}  def main(argv: list[str] | None = None) -> int:    p = argparse.ArgumentParser(description="Real-time ComfyUI WebSocket monitor")    p.add_argument("--host", default=DEFAULT_LOCAL_HOST, help="ComfyUI server URL")    p.add_argument("--api-key", help=f"API key for cloud (or set ${ENV_API_KEY} env var)")    p.add_argument("--client-id", default=None, help="Client ID (default: random UUID)")    p.add_argument("--prompt-id", default=None,                   help="Filter to a specific prompt_id (default: all jobs)")    p.add_argument("--previews", default=None,                   help="Directory to save in-progress preview frames")    p.add_argument("--no-color", action="store_true", help="Disable ANSI colour")    p.add_argument("--timeout", type=float, default=600.0,                   help="Hard cap on monitor duration (default 600s)")    args = p.parse_args(argv)     try:        import websocket  # type: ignore[import-not-found]    except ImportError:        print(json.dumps({            "error": "websocket-client not installed",            "install": "pip install websocket-client",        }))        return 1     api_key = resolve_api_key(args.api_key)    cloud = is_cloud_host(args.host)    client_id = args.client_id or new_client_id()     # Build WS URL preserving any base-path component (e.g. behind reverse proxy).    parsed = urlparse(args.host if "://" in args.host else f"http://{args.host}")    scheme = "wss" if parsed.scheme == "https" else "ws"    netloc = parsed.netloc    base_path = parsed.path.rstrip("/")    ws_url = f"{scheme}://{netloc}{base_path}/ws?clientId={client_id}"    if cloud and api_key:        ws_url += f"&token={api_key}"     color_on = not args.no_color and sys.stdout.isatty()     preview_dir = Path(args.previews).expanduser() if args.previews else None    if preview_dir:        preview_dir.mkdir(parents=True, exist_ok=True)        log(f"Saving previews to {preview_dir}")     log(f"Connecting to {ws_url} (client_id={client_id})")    if args.prompt_id:        log(f"Filtering messages to prompt_id={args.prompt_id}")     ws = websocket.create_connection(ws_url, timeout=args.timeout)    ws.settimeout(args.timeout)     preview_counter = 0    try:        while True:            try:                msg = ws.recv()            except websocket.WebSocketTimeoutException:                log(f"Idle for {args.timeout}s — exiting")                return 0            if isinstance(msg, bytes):                parsed = parse_binary_frame(msg)                if parsed is None:                    continue                if parsed["kind"] in {"preview", "preview_with_metadata"} and preview_dir:                    img_bytes = parsed.get("image_bytes", b"")                    if img_bytes:                        ext = parsed.get("ext", "png")                        out = preview_dir / f"preview_{preview_counter:05d}.{ext}"                        out.write_bytes(img_bytes)                        preview_counter += 1                        log(f"  [preview] saved {out.name} ({len(img_bytes)} bytes)")                continue             try:                payload = json.loads(msg)            except Exception:                continue            mtype = payload.get("type", "")            mdata = payload.get("data", {}) or {}            pid = mdata.get("prompt_id")             if args.prompt_id and pid and pid != args.prompt_id:                continue             if mtype == "status":                qr = mdata.get("status", {}).get("exec_info", {}).get("queue_remaining", "?")                print(fmt_color(f"[status] queue_remaining={qr}", DIM, color_on=color_on))            elif mtype == "execution_start":                print(fmt_color(f"[start] prompt_id={pid}", BOLD, color_on=color_on))            elif mtype == "executing":                node = mdata.get("node")                if node:                    print(fmt_color(f"  [executing] node={node}", CYAN, color_on=color_on))                else:                    print(fmt_color(f"  [executing] (workflow done) prompt_id={pid}", DIM, color_on=color_on))            elif mtype == "progress":                v, m = mdata.get("value", 0), mdata.get("max", 0)                pct = (v / m * 100) if m else 0                print(f"    [progress] {v}/{m} ({pct:5.1f}%) node={mdata.get('node')}")            elif mtype == "progress_state":                # Newer extended progress message                nodes = mdata.get("nodes") or {}                running = [k for k, v in nodes.items() if v.get("running")]                if running:                    print(fmt_color(f"    [progress_state] running={running}", DIM, color_on=color_on))            elif mtype == "executed":                node = mdata.get("node")                out = mdata.get("output") or {}                summary_parts = []                for key in ("images", "video", "videos", "gifs", "audio", "files"):                    if out.get(key):                        summary_parts.append(f"{key}={len(out[key])}")                summary = ", ".join(summary_parts) if summary_parts else "(no files)"                print(fmt_color(f"  [executed] node={node} {summary}", GREEN, color_on=color_on))            elif mtype == "execution_cached":                cached = mdata.get("nodes") or []                if cached:                    print(fmt_color(f"  [cached] {len(cached)} nodes skipped", DIM, color_on=color_on))            elif mtype == "execution_success":                print(fmt_color(f"[success] prompt_id={pid}", GREEN + BOLD, color_on=color_on))                if args.prompt_id:                    return 0            elif mtype == "execution_error":                exc_type = mdata.get("exception_type", "?")                exc_msg = mdata.get("exception_message", "?")                print(fmt_color(f"[error] {exc_type}: {exc_msg}", RED + BOLD, color_on=color_on))                tb = mdata.get("traceback")                if tb:                    if isinstance(tb, list):                        for line in tb:                            print(fmt_color(f"  {line}", RED, color_on=color_on))                    else:                        print(fmt_color(f"  {tb}", RED, color_on=color_on))                if args.prompt_id:                    return 1            elif mtype == "execution_interrupted":                print(fmt_color(f"[interrupted] prompt_id={pid}", YELLOW, color_on=color_on))                if args.prompt_id:                    return 1            elif mtype == "notification":                v = mdata.get("value", "")                print(fmt_color(f"[notification] {v}", DIM, color_on=color_on))            else:                # Unknown / lightly-used types: print compactly                print(fmt_color(f"[{mtype}] {json.dumps(mdata, default=str)[:200]}", DIM, color_on=color_on))     except KeyboardInterrupt:        log("Interrupted")        return 130    finally:        try:            ws.close()        except Exception:            pass  if __name__ == "__main__":    sys.exit(main())
← All Hermes Agent skillsProject instructions →