ComfyUI APIをPythonで操作する手順｜画像生成の自動化

ComfyUI には起動直後から使える HTTP API と WebSocket API が組み込まれている。Python からワークフロー JSON を送り、完了通知を受け取って画像を保存するまでの手順をまとめた。Stable Diffusion WebUI の --api オプションに慣れた人でも、ComfyUI の API 体系は別物なのでエクスポート方法から確認してほしい。

ComfyUI API の仕組みと SD WebUI との違い

まず構造上の違いを押さえておく。

ComfyUI の API は、GUI で組んだワークフロー全体を JSON として送信する設計になっている。「どのノードにどの値を渡すか」をすべて JSON に含めるため、ノード構成を変えるだけで txt2img も img2img も ControlNet も同じエンドポイントで扱える。

主要なエンドポイントは次の 3 つ。

WebSocket は ws://host:8188/ws?clientId=クライアントID で接続し、実行状態をリアルタイムに受信できる。

ワークフローを API 形式でエクスポートする

ComfyUI の GUI で組んだワークフローをそのまま API に送ることはできない。「API 形式」の JSON を取り出す手順：

1. ComfyUI（デフォルト http://localhost:8188）をブラウザで開く
2. 右上の歯車アイコン → 「Enable Dev mode Options」 をオンにする
3. ページをリロードする（設定が即反映されないことがある）
4. ワークフローを組み終えたら、ツールバーの 「Save (API Format)」 をクリック
5. workflow_api.json がダウンロードされる

「Save (API Format)」が見つからなければ Dev mode が有効になっていない可能性が高い。

エクスポートした JSON の構造は次のとおり。キーはノード ID（文字列の数字）で、inputs に各パラメータが入る。

{
  "3": {
    "class_type": "KSampler",
    "inputs": {
      "seed": 42,
      "steps": 20,
      "cfg": 7,
      "sampler_name": "euler",
      "scheduler": "normal",
      "denoise": 1,
      "model": ["4", 0],
      "positive": ["6", 0],
      "negative": ["7", 0],
      "latent_image": ["5", 0]
    }
  },
  "6": {
    "class_type": "CLIPTextEncode",
    "inputs": {
      "text": "a cat",
      "clip": ["4", 1]
    }
  }
}

Python からパラメータを変えるときは、このノード ID とキー名を直接書き換える。同じ class_type が複数ある場合（CLIPTextEncode など）は inputs.text の初期値や接続先のノード ID で判別する。

Python でプロンプトをキューに積む

標準ライブラリの urllib・json・uuid と、WebSocket 用の websocket-client（pip install websocket-client）で動く。

import json
import urllib.request
import uuid

SERVER = '127.0.0.1:8188'  # ComfyUI のアドレス

def queue_prompt(workflow: dict, client_id: str) -> str:
    payload = json.dumps({'prompt': workflow, 'client_id': client_id}).encode()
    req = urllib.request.Request(
        f'http://{SERVER}/prompt',
        data=payload,
        headers={'Content-Type': 'application/json'},
    )
    with urllib.request.urlopen(req) as res:
        data = json.loads(res.read())
    return data['prompt_id']

ワークフロー JSON を読み込み、変えたい値を書き換えてからキューに積む：

with open('workflow_api.json') as f:
    workflow = json.load(f)

# ノード ID は自分のワークフローに合わせて変更する
workflow['6']['inputs']['text'] = 'a cat sitting on a wooden chair, photorealistic'
workflow['7']['inputs']['text'] = 'blurry, low quality, watermark'
workflow['3']['inputs']['seed'] = 123456789

client_id = str(uuid.uuid4())
prompt_id = queue_prompt(workflow, client_id)
print(f'キュー登録完了: {prompt_id}')

ノード ID はワークフローによって変わる。送信前に次のコマンドで class_type を確認しておくと効率的。

grep -n 'class_type' workflow_api.json

WebSocket で完了を待ち、画像を取得する

プロンプトをキューに積んだ後、WebSocket で完了を待つ。node が null になったタイミングが全ノードの実行完了を表す。

import websocket

def wait_and_get_filenames(prompt_id: str, client_id: str) -> list:
    ws = websocket.WebSocket()
    ws.settimeout(300)  # 秒。長い生成ジョブでは延ばす
    ws.connect(f'ws://{SERVER}/ws?clientId={client_id}')
    try:
        while True:
            msg = json.loads(ws.recv())
            if msg['type'] == 'executing':
                data = msg['data']
                if data['node'] is None and data['prompt_id'] == prompt_id:
                    break  # 全ノード完了
    finally:
        ws.close()

    # 履歴からファイル名を取得
    with urllib.request.urlopen(
        f'http://{SERVER}/history/{prompt_id}'
    ) as res:
        history = json.loads(res.read())

    filenames = []
    for node_output in history[prompt_id]['outputs'].values():
        for img in node_output.get('images', []):
            filenames.append(img['filename'])
    return filenames

ファイル名が取得できたら /view エンドポイントでダウンロードする：

import urllib.parse
from pathlib import Path

def download_image(filename: str, save_dir: str = '.') -> Path:
    params = urllib.parse.urlencode({'filename': filename, 'type': 'output'})
    url = f'http://{SERVER}/view?{params}'
    save_path = Path(save_dir) / filename
    urllib.request.urlretrieve(url, save_path)
    return save_path

# 実行例
filenames = wait_and_get_filenames(prompt_id, client_id)
for fn in filenames:
    path = download_image(fn, save_dir='./outputs')
    print(f'保存: {path}')

FLUX や LoRA を組み込んだワークフローでも同じコードがそのまま使える。ワークフロー側の構成については ComfyUIでFLUX LoRAを適用する手順とトラブル対処 を参考にしてほしい。

つまずきやすいポイントと対処

Connection refused になる
ComfyUI はデフォルトで 127.0.0.1 のみリッスンしている。別マシンから叩くと繋がらない。--listen 0.0.0.0 で起動するか、SSH トンネルを経由する方が安全。

ssh -L 8188:localhost:8188 user@server-ip

その後は SERVER = '127.0.0.1:8188' のまま Python からアクセスできる。

WebSocket が途中で切れる
ws.settimeout(300) の値が足りない場合は延ばす。ControlNet を多段に重ねた構成や高解像度生成では、タイムアウトが発生しやすい場合があります。完了確認を /queue エンドポイントのポーリングに切り替える方法もある（環境によって異なります）。

ノード ID が分からない
GUI でノードをクリックすると右上隅にノード ID が表示される。JSON の class_type と照らし合わせると確実。同種ノードが複数ある場合は、接続先（["ノードID", 出力番号] の配列表記）をたどって判別する。

node_errors が返ってくる
queue_prompt のレスポンスに node_errors キーがあり、中身が空でない場合はワークフローの検証エラー。モデルファイルが見つからない・LoRA のパスが間違っているケースが代表的です。

まとめ

ComfyUI API と Python を繋ぐ手順をまとめると次のようになる。

SD WebUI API と比べると初期設定にひと手間あるが、ワークフロー全体を JSON で送る設計のおかげで、ノード構成を変えてもコードを書き直さずに済む。まず手元の ComfyUI でシンプルな KSampler ワークフローを API 形式でエクスポートし、上のコードをそのまま実行してみてほしい。