ComfyUI API 教學：用 Python 批次產圖，無須操作介面

1{
2  "9": {
3    "inputs": { "filename_prefix": "moss_image_t2i", "images": ["18:8", 0] },
4    "class_type": "SaveImage"
5  },
6  "18:7": {
7    "inputs": { "seed": 438829726369282, "steps": 25, "cfg": 2.5 },
8    "class_type": "KSampler"
9  },
10  "18:17": {
11    "inputs": { "value": "Cinematic night street photograph: ..." },
12    "class_type": "PrimitiveStringMultiline"
13  }
14}

1import json
2import pathlib
3import urllib.error
4import urllib.parse
5import urllib.request
6import uuid
7
8import websocket
9
10SERVER = "127.0.0.1:8188"
11OUTPUT_DIR = pathlib.Path("output")
12
13POSTS = [
14    {"name": "mochi", "seed": 20261103, "prompt": "貓咪咖啡廳的 Instagram 店員介紹貼文，3:4 直式。上方寫「本週店員」，中間是一隻戴小圍裙的橘貓照片，下方大字寫名字「麻糬」，名字下面寫職稱「店長」。暖色、溫馨的咖啡廳氛圍，版面乾淨。"},
15    {"name": "inky", "seed": 20261104, "prompt": "貓咪咖啡廳的 Instagram 店員介紹貼文，3:4 直式。上方寫「本週店員」，中間是一隻戴小圍裙的黑貓照片，下方大字寫名字「墨墨」，名字下面寫職稱「咖啡師」。暖色、溫馨的咖啡廳氛圍，版面乾淨。"},
16    {"name": "bean", "seed": 20261105, "prompt": "貓咪咖啡廳的 Instagram 店員介紹貼文，3:4 直式。上方寫「本週店員」，中間是一隻戴小領結的賓士貓照片，下方大字寫名字「豆豆」，名字下面寫職稱「保全」。暖色、溫馨的咖啡廳氛圍，版面乾淨。"},
17]
18
19
20def queue_workflow(workflow, client_id):
21    body = json.dumps({"prompt": workflow, "client_id": client_id}).encode()
22    request = urllib.request.Request(
23        f"http://{SERVER}/prompt",
24        data=body,
25        headers={"Content-Type": "application/json"},
26    )
27    try:
28        with urllib.request.urlopen(request) as response:
29            result = json.load(response)
30    except urllib.error.HTTPError as error:
31        raise RuntimeError(error.read().decode()) from error
32    if result["node_errors"]:
33        raise RuntimeError(result["node_errors"])
34    return result["prompt_id"]
35
36
37def wait_until_done(ws, prompt_id):
38    while True:
39        message = ws.recv()
40        if isinstance(message, bytes):
41            continue  # 開啟預覽時，取樣中的預覽圖會以二進位傳來
42        event = json.loads(message)
43        data = event.get("data", {})
44        if data.get("prompt_id") != prompt_id:
45            continue
46        if event["type"] == "execution_error":
47            raise RuntimeError(data.get("exception_message"))
48        if event["type"] == "executing" and data.get("node") is None:
49            return
50
51
52def download_images(prompt_id, name):
53    with urllib.request.urlopen(f"http://{SERVER}/history/{prompt_id}") as response:
54        outputs = json.load(response)[prompt_id]["outputs"]
55    for node_output in outputs.values():
56        for index, image in enumerate(node_output.get("images", [])):
57            if image["type"] != "output":
58                continue
59            query = urllib.parse.urlencode(
60                {"filename": image["filename"], "subfolder": image["subfolder"], "type": image["type"]}
61            )
62            with urllib.request.urlopen(f"http://{SERVER}/view?{query}") as response:
63                path = OUTPUT_DIR / f"{name}_{index}.png"
64                path.write_bytes(response.read())
65                print(f"{name}: {path}")
66
67
68workflow = json.loads(pathlib.Path("cat-post-api.json").read_text(encoding="utf-8"))
69OUTPUT_DIR.mkdir(exist_ok=True)
70client_id = str(uuid.uuid4())
71ws = websocket.create_connection(f"ws://{SERVER}/ws?clientId={client_id}")
72
73for post in POSTS:
74    workflow["18:17"]["inputs"]["value"] = post["prompt"]
75    workflow["18:7"]["inputs"]["seed"] = post["seed"]
76    workflow["9"]["inputs"]["filename_prefix"] = f"cats/{post['name']}"
77    prompt_id = queue_workflow(workflow, client_id)
78    wait_until_done(ws, prompt_id)
79    download_images(prompt_id, post["name"])
80
81ws.close()

1uv run --with websocket-client generate_cat_posts.py