tools/run_workflow.py

"""Run an included API workflow against a local ComfyUI server."""
import argparse
import json
from pathlib import Path
import time
from urllib.request import Request, urlopen

parser = argparse.ArgumentParser()
parser.add_argument('workflow', type=Path)
parser.add_argument('--server', default='http://127.0.0.1:8188')
args = parser.parse_args()
graph = json.loads(args.workflow.read_text(encoding='utf-8'))
request = Request(args.server + '/prompt', data=json.dumps({'prompt':graph}).encode(), headers={'Content-Type':'application/json'})
with urlopen(request, timeout=30) as response:
    prompt_id = json.load(response)['prompt_id']
print('Queued:',prompt_id,flush=True)
while True:
    with urlopen(args.server + '/history/' + prompt_id, timeout=30) as response:
        history = json.load(response).get(prompt_id)
    if history:
        print(json.dumps(history['status'],indent=2))
        print(json.dumps(history['outputs'],indent=2))
        if history['status']['status_str'] != 'success':
            raise SystemExit(1)
        break
    time.sleep(0.5)