ComfyUI APIで画像を一括生成する方法：ワークフロー自動化

安定して動くComfyUIワークフローがあり、200商品それぞれに4枚のメイン画像を作りたいとします。画面の前でpromptとseedを毎回変更する作業は現実的ではありません。ところが、最初にスクリプトから/promptへPOSTするとnode_errorsが返り、出力したJSONがAPI formatではなかったことに気づく場合があります。処理が終わっても、/historyにはfilenameとsubfolderしかなく、実ファイルは/viewから取得する必要があります。

GUIからスクリプトによる量産へ移る際の要点は3つです。正しいAPI formatのワークフローを出力すること、無目的なポーリングではなくWebSocketで完了を待つこと、OOMを避けるため同時実行数とVRAMを制御することです。以下では最小スクリプト、パラメータ化、一括キュー、バックエンド向けの設計を順に扱います。

ローカルComfyUI Server APIの入口

CLI起動とデフォルトポート

ComfyUIのローカルサービスは、デフォルトで127.0.0.1:8188をリッスンします。同じ端末内から使うだけなら通常起動で十分です。LAN内の別端末から接続するときだけ、リッスンアドレスを明示します。

# ローカルからのみアクセス
python main.py --port 8188

# LANアクセスを許可する。ファイアウォール、認証、またはリバースプロキシも設定する
python main.py --listen 0.0.0.0 --port 8188

ターミナルにStarting serverと表示されたら、http://127.0.0.1:8188を開きます。ComfyUIのフロントエンドが表示されれば準備完了です。

VRAMが不足する場合は、現在のバージョンでサポートされている--lowvram、--novram、最後の手段として低速な--cpuを検討します。余裕がある場合は--highvramもテストできます。GPU容量だけで機械的に選ばないでください。モデル、精度、VAE、ワークフロー構成によってピークは変わります。python main.py --helpと最新の公式トラブルシューティングを確認します。

主要API Routes一覧

ローカルComfyUI Serverの主なルートはserver.pyに定義されています。スクリプトでよく使うエンドポイントは次のとおりです。

ルート	用途	パラメータ／戻り値
/prompt	workflowを検証してキューへ追加	POST {"prompt": workflow_dict, "client_id": "..."}。成功時はprompt_id、失敗時はnode_errors
/history/{prompt_id}	実行履歴と出力metadataを取得	GET。filename、subfolder、typeを含むoutputs
/view?filename=...&subfolder=...&type=...	出力ファイルを取得	GET。バイナリデータ
/ws	WebSocketで実行状態を受信	ws://127.0.0.1:8188/ws?clientId=...
/queue	現在のキューを確認	GET。待機中ジョブの一覧
/interrupt	現在の実行を中断	POST。タイムアウト時に使用
/upload/image	画像入力をアップロード	POST multipart/form-data
/object_info	使用可能なノード型とパラメータを取得	GET。ノードの有無を確認するときに使用

ルートはComfyUIのバージョンによって追加・変更される可能性があります。挙動が違う場合は最新の公式文書またはserver.pyを確認してください。

WebSocketメッセージの種類

ジョブの完了を待つときは、次のメッセージを監視します。

• status：queue_remainingを含むキュー状態
• execution_start：prompt_idを伴う実行開始
• execution_cached：キャッシュから再利用されたノード
• executing：現在の実行ノード。node is Noneでprompt_idが一致すれば完了
• progress：現在のステップ数と総ステップ数
• executed：ノード完了と出力metadata

完了判定では、type == "executing"のメッセージを受け取り、data.nodeがNoneで、data.prompt_idが送信時のprompt_idと一致することを確認します。

API Format Workflowを出力する

Export Workflow (API)の手順

ComfyUIフロントエンドが保存するworkflow JSONと、APIが受け取る形式は異なります。通常のworkflowをそのままPOSTすると、node_errorsで検証に失敗することがあります。

出力手順は次のとおりです。

1. ComfyUIフロントエンドで、正常に画像を生成できるワークフローを読み込む
2. File -> Export Workflow (API)を選ぶ。バージョンによってはSave (API Format)と表示されるため、現在のUIで同等の項目を使う
3. workflow_api.jsonなどの.jsonファイルとして保存する
4. JSONを開き、"3"や"6"のような数値node IDがあり、各ノードにclass_typeとinputsが含まれることを確認する

フロントエンド更新でメニュー名が変わる可能性があるため、使用中のバージョンでAPI出力に相当する機能を選んでください。

API formatとSave formatの違い

通常のSave formatにはフロントエンドのレイアウト情報が含まれます。API formatはそれを除き、実行に必要な情報を保持します。

形式	含まれる内容	用途
Save format	ノード位置、色、グループ、サイズ、リンク表示情報	フロントエンドでレイアウトを再編集
API format	数値node ID、class_type、inputs、任意の_meta	スクリプトやAPIから実行

Save formatのJSONを/promptへ直接送ると、node_errorsまたはerrorが返ることがあります。フロントエンドで読み込み、API formatとして再出力します。

Node IDを管理する

パラメータ化にはprompt、seed、画像サイズ、出力ノードのIDが必要です。出力したJSONを開き、次を探します。

• Prompt node：CLIPTextEncode。単純な例では"6"の場合がある
• Seed node：KSampler。単純な例では"3"の場合がある
• Width/Height node：EmptyLatentImageなど、実際のワークフローに応じたノードのinputs
• Output node：SaveImageまたはSaveImageWebsocket

フロントエンドのNoteやGroupで用途を残すと追跡しやすくなります。ただしスクリプトでは、出力されたJSONを確認する必要があります。IDはグラフごとに生成され、再出力で変わることがあるため、固定値だと思わないでください。

最小スクリプト：送信、待機、取得

API呼び出しは、workflowを/promptへ送信し、完了を待ち、/historyからmetadataを取得して/viewからファイルを取得する3段階です。

HTTPで送信だけ行う

最小構成では/promptへPOSTし、完了を待ちません。別のworkerがジョブを確認する構成に向いています。

import json
import requests

# API formatのworkflowを読み込む
with open("workflow_api.json", "r") as f:
    workflow = json.load(f)

# リクエストデータを作成する
payload = {
    "prompt": workflow,
    "client_id": "my-script-client"  # 任意。WebSocketイベントとの関連付けに使う
}

# キューへ送信する
response = requests.post("http://127.0.0.1:8188/prompt", json=payload)

if response.status_code == 200:
    result = response.json()
    prompt_id = result["prompt_id"]
    print(f"送信しました: {prompt_id}")
else:
    error = response.json()
    print(f"送信に失敗しました: {error}")
    # node_errorsにノード単位の検証内容が入る

成功時は{"prompt_id": "...", "number": ...}、失敗時は{"error": {...}, "node_errors": {...}}が返ります。送信はキューへの追加であり、完了待ちではありません。

WebSocketで完了を待つ

WebSocketを使えば、短い間隔でポーリングする必要はありません。先に接続し、ジョブを送信してからexecutingの終了イベントを待ちます。

import json
import uuid
import requests
import websocket

# workflowを読み込む
with open("workflow_api.json", "r") as f:
    workflow = json.load(f)

# client_idを生成する
client_id = str(uuid.uuid4())

# WebSocketへ接続する
ws = websocket.create_connection(f"ws://127.0.0.1:8188/ws?clientId={client_id}")

# ジョブを送信する
payload = {"prompt": workflow, "client_id": client_id}
response = requests.post("http://127.0.0.1:8188/prompt", json=payload)
prompt_id = response.json()["prompt_id"]

# 完了を待つ
while True:
    message = ws.recv()
    data = json.loads(message)

    if data["type"] == "executing":
        # nodeがNoneでprompt_idが一致すれば完了
        if data["data"]["node"] is None and data["data"]["prompt_id"] == prompt_id:
            print("ジョブが完了しました")
            break

ws.close()

300秒などの最大待機時間を追加してください。/interruptは現在の実行を止めるため、タイムアウト時も対象を確認して使います。

HistoryとViewで画像を取得する

完了後に/history/{prompt_id}から出力metadataを取得し、各ファイルを/viewからダウンロードします。

import requests

# ジョブ履歴を取得する
history_url = f"http://127.0.0.1:8188/history/{prompt_id}"
history = requests.get(history_url).json()

# 出力ノードを調べる
outputs = history[prompt_id]["outputs"]
for node_id, node_output in outputs.items():
    if "images" in node_output:
        for image in node_output["images"]:
            filename = image["filename"]
            subfolder = image.get("subfolder", "")
            type = image.get("type", "output")

            # ダウンロードURLを作る
            view_url = f"http://127.0.0.1:8188/view?filename={filename}&subfolder={subfolder}&type={type}"

            # バイナリ画像を保存する
            img_data = requests.get(view_url).content
            with open(f"output_{filename}", "wb") as f:
                f.write(img_data)
            print(f"保存しました: output_{filename}")

outputsは{node_id: {"images": [{"filename": "...", "subfolder": "...", "type": "..."}]}}という構造です。historyはmetadataを返し、/viewが画像のバイナリを返します。

一括処理のパラメータ化

安定したworkflowをテンプレートにし、各反復でprompt、seed、画像サイズなど選んだ入力だけを変更します。

Prompt、Seed、画像サイズを変数化する

各ノードのinputsにある値を変更します。

import json
import random

# workflowを読み込む
with open("workflow_api.json", "r") as f:
    workflow = json.load(f)

# promptを変更する。IDは実際のworkflowに合わせる
workflow["6"]["inputs"]["text"] = "a beautiful landscape, sunset, mountains"

# ランダムseedを生成する
workflow["3"]["inputs"]["seed"] = random.randint(0, 1000000)

# 画像サイズを変更する
workflow["5"]["inputs"]["width"] = 1024
workflow["5"]["inputs"]["height"] = 768

# ジョブを送信する...

"6"、"3"、"5"は、実際に出力したworkflowで確認する必要があります。prompt一覧をループして、送信ごとにpromptとseedを変更できます。

prompts = [
    "product photo, white background",
    "product photo, outdoor scene",
    "product photo, studio lighting"
]

for i, prompt_text in enumerate(prompts):
    workflow["6"]["inputs"]["text"] = prompt_text
    workflow["3"]["inputs"]["seed"] = random.randint(0, 1000000)

    # ジョブを送信する
    payload = {"prompt": workflow, "client_id": client_id}
    response = requests.post("http://127.0.0.1:8188/prompt", json=payload)
    prompt_id = response.json()["prompt_id"]

    # WebSocketで待つ...
    # 出力を取得する...

公式サンプルでもKSampler.seedとCLIPTextEncode.textを変更しており、パラメータの位置を探す出発点になります。

一括キューの戦略

200枚を生成する場合、workflow内のbatch sizeを増やす方法と、複数promptを送信する方法があります。モデルと実測したVRAM余裕によって選びます。

戦略	特徴	適する状況
1送信につき1枚	ジョブごとのVRAMを制御しやすい	VRAM余裕が少ない、大きなモデル、複雑なworkflow
Batch size	1つのprompt内で複数枚を生成し、通常はVRAMピークが高い	実測上の余裕が大きい、小さなモデル、単純なworkflow
同時リクエスト	複数promptを送り、実行中の数を制限	容量を測定済みでworkerを分離できる場合

20件を一度に送るとキューが膨らみ、OOMまたはサービス停止につながる可能性があります。

安全な初期値は、直列送信、WebSocketによる完了待ち、VRAM監視です。前のジョブが終わってから次を送ります。メモリ不足が続くなら、同じ処理の再試行ではなく負荷を下げ、現在の低メモリ設定を検証します。

同時実行を制御する例

複数GPUまたは分離したクラウドworkerでは、Semaphoreで実行数を制限できます。

import copy
import threading

# 同時に最大2件まで実行する
semaphore = threading.Semaphore(2)

def submit_and_wait(prompt_text, seed):
    with semaphore:
        # 共有workflowを変更せず、タスクごとに独立した複製を使う
        job_workflow = copy.deepcopy(workflow)
        job_workflow["6"]["inputs"]["text"] = prompt_text
        job_workflow["3"]["inputs"]["seed"] = seed

        # 送信して完了を待つ...
        # WebSocketループ...

    # permitが返され、次のタスクが進める

# 一括送信
threads = []
for i in range(50):
    t = threading.Thread(target=submit_and_wait, args=(prompts[i], seeds[i]))
    threads.append(t)
    t.start()

for t in threads:
    t.join()

同時実行数1から始め、実際のworkflowでメモリピーク、遅延、失敗率を測ってから増やします。同じVRAM容量でも、精度、VAE、後処理ノード、workerの分離方法で上限は変わります。

VRAMを監視する

一括実行中はシステム統計を取得し、しきい値を超えたら新規送信を待機させます。

import time

def check_vram(threshold=0.8):
    stats = requests.get("http://127.0.0.1:8188/system_stats").json()
    vram_used = stats["system_stats"]["devices"][0]["vram_used"]
    vram_total = stats["system_stats"]["devices"][0]["vram_total"]
    return (vram_used / vram_total) > threshold

# 送信前にVRAMを確認する
for prompt_text in prompts:
    while check_vram(0.85):
        print("VRAM使用率が高いため30秒待機します...")
        time.sleep(30)

    # 次のジョブを送信する...
    # WebSocketで完了を待つ...

メモリピークはKSampler実行中に発生しやすく、完了後に下がる場合があります。サービスへ過剰な要求を送らないよう、10〜30秒間隔で監視します。

一括出力を保存する

大量の出力には予測可能な配置が必要です。{prompt_id}_{seed}_{timestamp}.pngのように、タスクID、seed、生成時刻をファイル名へ含めます。

少なくとも次を記録します。

• prompt_id：ComfyUIのタスクID
• seed：乱数seed
• prompt_text：使用したprompt
• width/height：出力サイズ
• timestamp：生成時刻
• business_id：商品SKUや注文番号などの業務ID

outputs/20260624/batch_001/のように日付またはバッチ単位で整理します。ファイルだけで追跡しにくくなったら、SQLiteやPostgreSQLでmetadataと画像パスを関連付けます。

バックエンド連携の設計チェック

バックエンドで包む場合は、request ID、タイムアウト、キュー上限、VRAM監視、エラー処理が必要です。動くスクリプトだけでは安定したサービスになりません。

Request IDと冪等性

UUIDや注文番号など一意の業務request_idを作り、ComfyUIのprompt_idと対応付けます。完了済みrequest_idが再度届いた場合は、再生成せず既存結果を返します。

import uuid

# 業務request ID
request_id = str(uuid.uuid4())

# データベースまたはキャッシュに対応を保存する
request_prompt_map[request_id] = prompt_id

# 重複リクエストには既存結果を返す
if request_id in completed_requests:
    return get_cached_result(request_id)

prompt_idはComfyUIが生成するため、業務request_idとは別物です。対応関係を自分で永続化します。

タイムアウトとキャンセル

1ジョブの最大時間を300秒などに設定します。超過したら/interruptで現在の実行を止め、キューを進めます。

import time

timeout = 300  # 5分
start_time = time.time()

# WebSocketメッセージを待つ...
while True:
    elapsed = time.time() - start_time
    if elapsed > timeout:
        # 現在の実行を中断する
        requests.post("http://127.0.0.1:8188/interrupt")
        print("タイムアウトしたためジョブを中断しました")
        break

    # 通常のメッセージを処理する...

WebSocketが切断されたり一定時間応答がなかったりした場合は、再接続またはhistory照会へ切り替えます。中断後は/queueを確認し、待機中ジョブを残すか判断します。

キュー上限とVRAM監視

待機ジョブを最大5件にするなど上限を設け、超過時は拒否または後回しにします。同時worker数も制限し、複数要求が予期しないメモリピークを作らないようにします。

/system_statsでVRAMを確認します。

stats = requests.get("http://127.0.0.1:8188/system_stats").json()
vram_used = stats["system_stats"]["devices"][0]["vram_used"]
vram_total = stats["system_stats"]["devices"][0]["vram_total"]
vram_percent = vram_used / vram_total

if vram_percent > 0.8:
    print("VRAM使用率が高いため新規リクエストを拒否します")

負荷が高いときは新規要求を拒否するかキューが減るまで待ちます。同じ処理を再試行する前に、batch sizeを下げ、workflowを簡略化します。

エラー分類と再試行

エラーごとに対応を変えます。

エラー	主な原因	対応
node_errors	モデルやノード不足、無効なパラメータ、入力欠落	workflowと環境を修正し、再試行しない
OOM	VRAM不足	batch sizeや負荷を下げ、検証済みのメモリ設定を変更。同じ条件では再試行しない
WebSocket切断	ネットワークまたは接続の問題	再接続して/history/{prompt_id}を確認
ジョブのタイムアウト	モデル読み込みが遅い、workflowが複雑	タイムアウトを延ばすか簡略化し、再試行は最大1回
サービス停止	メモリ枯渇、GPU障害	ログを確認して再起動し、慎重に再試行

node_errorsには、存在しないノード型や入力型の不一致など、検証に失敗したノードが示されます。workflow、custom nodes、モデルファイルを先に修正します。OOMも同じ要求の即時再試行では解決しません。

Cloud APIとローカルAPIの比較

Comfy Cloudはホストされたワークフロー実行を提供しますが、認証、ジョブ状態、WebSocket URL、同時実行制限はローカルServer APIと異なります。API formatの考え方は共通でも、コードは最新のCloud API Referenceに合わせます。

ルートの違い

主なルートは次のように異なります。

機能	ローカルAPI	Cloud API
ジョブ送信	/prompt	/api/prompt
状態／結果確認	/history/{prompt_id}	/api/job/{prompt_id}/statusと/api/jobs/{job_id}
出力取得	/view	/api/view
WebSocket	/ws?clientId=...	/ws?clientId=...&token=...

CloudにはX-API-Keyヘッダーと有効なサブスクリプションが必要です。ローカルAPIはデフォルトで127.0.0.1だけをリッスンしますが、--listen、リバースプロキシ、ポート転送で外部公開するなら認証とアクセス制御を自分で実装します。Cloud APIは現在もexperimentalです。旧/api/history_v2/{prompt_id}はdeprecatedで、/api/jobs/{job_id}の利用が推奨されています。

同時実行とサブスクリプション制限

Cloudの同時実行数はサブスクリプション階層によって決まり、上限を超えたジョブはキューで待機します。出力はcloud storageに保存され、/api/viewは一時的な署名付きURLを返します。

プラン、同時実行数、実行時間、価格は変わるため、固定値は記載しません。CloudではローカルGPUの管理が不要ですが、ローカルServer APIではキュー、VRAM、セキュリティ、サービス安定性を自分で管理します。

次に読む内容

シリーズ内の記事

このページは、動作する画像ワークフローからプログラムによる一括生成とバックエンド連携へ進むための工程です。

• ComfyUIワークフロー再利用ガイド：workflowの読み込み、missing nodes、モデルパス
• ComfyUIの低VRAM・高速化：メモリ設定、batch、OOM、Tiled VAE
• ComfyUI動画生成：画像一括処理と動画固有workflowの境界
• ComfyUIのトラブルシューティングと保守：ノード不足、起動失敗、バージョン競合

関連テーマ

より広い自動化とバックエンド設計には、次の内容が役立ちます。

• n8nによるAIワークフロー自動化：ComfyUIと複数ツールの連携
• Ollama API実践：モデルのプログラム呼び出し、キュー、構造化出力
• LLMの構造化出力：信頼できるデータ抽出とAPI連携

まとめ

ComfyUIをGUIからスクリプトによる量産へ移す流れは、API formatのworkflowを出力し、WebSocketで完了を待ち、/historyと/viewからファイルを取得し、選んだ入力を変数化した後にrequest ID、タイムアウト、キュー上限、VRAM監視を追加することです。

まず1つのAPI format workflowを出力し、最小の送信・待機スクリプトを動かします。1枚が成功したら、パラメータを変えた10枚を生成してVRAMとキューの挙動を観察します。その小規模バッチが安定してから、冪等性、キャンセル、監視、結果記録を追加してください。