Function _run_comfy_workflow

tools/modly-cli/agent.py:546–571

(args: argparse.Namespace)

def _run_comfy_workflow(args: argparse.Namespace) -> dict[str, Any]:
    host = args.comfy_url.rstrip("/")
    workflow = _load_comfy_workflow(args.workflow, host=host, timeout=args.request_timeout)
    prompt = getattr(args, "prompt", None)
    seed = getattr(args, "seed", None)
    graph = _patch_comfy_workflow(workflow, prompt=prompt, seed=seed)
    payload = json.dumps({"prompt": graph, "client_id": "modly-cli"}).encode("utf-8")
    queued = _request_json("POST", f"{host}/prompt", timeout=args.request_timeout, data=payload, headers={"Content-Type": "application/json"})
    prompt_id = queued.get("prompt_id") if isinstance(queued, dict) else None
    if not prompt_id:
        raise ModlyCliError(f"ComfyUI did not return prompt_id: {queued}")

    deadline = time.monotonic() + args.timeout
    history: dict[str, Any] = {}
    while time.monotonic() < deadline:
        data = _request_json("GET", f"{host}/history/{urllib.parse.quote(str(prompt_id))}", timeout=args.request_timeout)
        if isinstance(data, dict) and str(prompt_id) in data:
            history = data[str(prompt_id)] if isinstance(data[str(prompt_id)], dict) else {"raw": data[str(prompt_id)]}
            break
        if getattr(args, "progress", False) and not getattr(args, "quiet", False):
            print(json.dumps({"phase": "comfy", "prompt_id": prompt_id, "status": "running"}), file=sys.stderr)
        time.sleep(args.poll)
    if not history:
        raise ModlyCliError(f"Timed out waiting for ComfyUI prompt {prompt_id}", code="TIMEOUT")

    return {"ok": True, "comfy_url": host, "workflow": args.workflow, "prompt_id": str(prompt_id), "history": history}


def _iter_comfy_file_refs(value: Any) -> list[dict[str, Any]]: