ComfyUI API on Ubuntu — PythonからWebSocket APIで画像生成を自動化

ComfyUI を使い込んでいると、「ボタンを押すのではなく、プログラムから画像生成を呼び出したい」という場面が出てきます。バッチ処理、別サービスとの連携、スケジュール実行など、理由は様々です。

ComfyUI には公式の Python SDK はありませんが、REST API と WebSocket API が最初から備わっています。websocket-client と requests があれば、数十行のコードでワークフローを送信し、生成完了まで待機して画像を受け取るクライアントが書けます。

本記事では Ubuntu 22.04 上に実際に ComfyUI 0.25.0 を起動し、Python から API に接続した結果を載せます。前提として ComfyUI の Ubuntu セットアップ が完了していることを想定しています。

ComfyUI API の全体像

ComfyUI の API は大きく2系統に分かれています。

• REST API（HTTP）：状態の確認、ワークフローの投入、履歴の参照に使う
• WebSocket（ws://）：生成の進捗をリアルタイムで受け取る専用チャンネル

画像1枚を自動生成する基本フローはこうなります。

1. WebSocket に接続し、自分のクライアント ID（UUID）を登録する
2. POST /prompt にワークフロー JSON を送る（このとき client_id を一緒に渡す）
3. WebSocket から executing・progress・executed の順にメッセージが届く
4. executed メッセージに含まれる prompt_id で GET /history/{prompt_id} を叩き、画像ファイル名を取得する
5. GET /view?filename=xxx&subfolder=&type=output で画像バイナリを取得する

Python クライアントの環境を作る

①必要パッケージをインストールする

Python の標準ライブラリだけでは WebSocket が扱いにくいため、websocket-client を追加します。画像のデコードに Pillow、非同期版を使うなら aiohttp も入れておくと後で役立ちます。

②接続テストで ComfyUI の状態を確認する

まず REST API だけで接続テストをします。/system_stats は ComfyUI のバージョンやデバイス情報を返します。正常に応答すれば、次のステップに進めます。

実際に私の環境で実行したところ、ComfyUI 0.25.0 で 792 種類のノードタイプが確認できました。また WebSocket への接続は即座に成功し、サーバーから queue_remaining: 0 という初期ステータスが届きました。

WebSocket 接続の仕組みを理解する

WebSocket 接続のコードを見てみましょう。ポイントは clientId クエリパラメータに UUID を渡すことです。後で POST /prompt に同じ UUID を渡すことで、「この WebSocket にこのジョブの進捗を流す」という紐付けが成立します。

import uuid
import websocket

client_id = str(uuid.uuid4())
ws = websocket.WebSocket()
ws.connect(f"ws://127.0.0.1:8188/ws?clientId={client_id}")
print("connected:", ws.connected)

# 接続直後にサーバーから status が届く
msg = ws.recv()
import json
print(json.loads(msg))
# -> {"type": "status", "data": {"status": {"exec_info": {"queue_remaining": 0}}, "sid": "..."}}

接続直後に受け取る status メッセージの data.sid フィールドがサーバー側のセッション ID です。通常は自分の client_id と一致しています。

ワークフロー JSON を POST して生成を開始する

①ワークフロー JSON の構造

ComfyUI の API では「ノードのグラフ」を JSON で表現します。GUI で組んだワークフローを [Save (API Format)] ボタンでエクスポートすると、以下のような形式になります。

{
  "3": {
    "class_type": "KSampler",
    "inputs": {
      "seed": 12345,
      "steps": 20,
      "cfg": 7,
      "sampler_name": "euler",
      "scheduler": "normal",
      "denoise": 1,
      "model": ["4", 0],
      "positive": ["6", 0],
      "negative": ["7", 0],
      "latent_image": ["5", 0]
    }
  },
  "4": {
    "class_type": "CheckpointLoaderSimple",
    "inputs": {"ckpt_name": "v1-5-pruned-emaonly.safetensors"}
  },
  "5": {
    "class_type": "EmptyLatentImage",
    "inputs": {"batch_size": 1, "height": 512, "width": 512}
  },
  "6": {
    "class_type": "CLIPTextEncode",
    "inputs": {"clip": ["4", 1], "text": "a beautiful landscape"}
  },
  "7": {
    "class_type": "CLIPTextEncode",
    "inputs": {"clip": ["4", 1], "text": "ugly, blurry"}
  },
  "8": {
    "class_type": "VAEDecode",
    "inputs": {"vae": ["4", 2], "samples": ["3", 0]}
  },
  "9": {
    "class_type": "SaveImage",
    "inputs": {"filename_prefix": "ComfyUI", "images": ["8", 0]}
  }
}

各ノードはオブジェクト ID（文字列の数字）をキーに持ちます。"inputs" 内の配列（例：["4", 0]）は「ノード 4 の出力 0 番を使う」という接続を表します。ComfyUI が扱う 792 種類のノードの入出力スキーマは GET /object_info で全取得できます。

②POST /prompt で送信する

import requests, json, uuid

COMFYUI_URL = "http://127.0.0.1:8188"
client_id = str(uuid.uuid4())

workflow = { ... }  # 上記のワークフロー JSON

payload = {
    "prompt": workflow,
    "client_id": client_id,
}
r = requests.post(f"{COMFYUI_URL}/prompt", json=payload)
result = r.json()

prompt_id = result["prompt_id"]
print("prompt_id:", prompt_id)
# -> prompt_id: f47ac10b-58cc-4372-a567-0e02b2c3d479

WebSocket で進捗を受け取り、画像を保存する

完全な自動化スクリプト

以下は、ワークフロー送信から画像保存までを一気通貫で行うスクリプトです。実際に使う場合は workflow_api.json に自分のワークフローを用意してください。

#!/usr/bin/env python3
"""
comfyui_client.py
ComfyUI WebSocket API を使って画像生成を自動化するスクリプト。
依存: pip install websocket-client requests Pillow
"""
import json
import uuid
import io
import websocket
import requests
from PIL import Image

COMFYUI_URL = "http://127.0.0.1:8188"


def queue_prompt(workflow: dict, client_id: str) -> str:
    """ワークフローをキューに追加し、prompt_id を返す"""
    payload = {"prompt": workflow, "client_id": client_id}
    r = requests.post(f"{COMFYUI_URL}/prompt", json=payload)
    r.raise_for_status()
    return r.json()["prompt_id"]


def wait_for_completion(ws: websocket.WebSocket, prompt_id: str) -> None:
    """WebSocket から executing/progress/executed メッセージを受け取り、完了まで待機する"""
    while True:
        msg = ws.recv()

        # バイナリメッセージ（プレビュー画像）はスキップ
        if isinstance(msg, bytes):
            continue

        data = json.loads(msg)
        msg_type = data.get("type")

        if msg_type == "executing":
            node_id = data["data"].get("node")
            if node_id is None and data["data"].get("prompt_id") == prompt_id:
                # 全ノードの実行が終わった
                print("Generation complete!")
                break
            print(f"  Executing node: {node_id}")

        elif msg_type == "progress":
            step  = data["data"]["value"]
            total = data["data"]["max"]
            print(f"  Progress: {step}/{total}")

        elif msg_type == "execution_error":
            raise RuntimeError(f"Execution error: {data['data']}")


def get_output_images(prompt_id: str) -> list[Image.Image]:
    """履歴から出力画像を取得して PIL.Image リストで返す"""
    r = requests.get(f"{COMFYUI_URL}/history/{prompt_id}")
    history = r.json()

    images = []
    outputs = history[prompt_id]["outputs"]
    for node_id, node_output in outputs.items():
        for img_info in node_output.get("images", []):
            params = {
                "filename": img_info["filename"],
                "subfolder": img_info["subfolder"],
                "type": img_info["type"],
            }
            img_r = requests.get(f"{COMFYUI_URL}/view", params=params)
            images.append(Image.open(io.BytesIO(img_r.content)))
    return images


def main():
    client_id = str(uuid.uuid4())

    # ワークフロー JSON を読み込む
    with open("workflow_api.json") as f:
        workflow = json.load(f)

    # WebSocket に接続
    ws = websocket.WebSocket()
    ws.connect(f"ws://127.0.0.1:8188/ws?clientId={client_id}")
    print("WebSocket connected")

    try:
        # ジョブを投入
        prompt_id = queue_prompt(workflow, client_id)
        print(f"Queued: prompt_id={prompt_id}")

        # 完了まで待機
        wait_for_completion(ws, prompt_id)

        # 画像を取得・保存
        images = get_output_images(prompt_id)
        for i, img in enumerate(images):
            out_path = f"output_{i:03d}.png"
            img.save(out_path)
            print(f"Saved: {out_path} ({img.size[0]}x{img.size[1]}px)")
    finally:
        ws.close()


if __name__ == "__main__":
    main()

非同期版（aiohttp + websockets）

複数ジョブを並列投入したい場合や、非同期フレームワーク（FastAPI 等）と組み合わせる場合は、aiohttp と websockets ライブラリの非同期版が向いています。

import asyncio
import json
import uuid
import aiohttp

COMFYUI_URL = "http://127.0.0.1:8188"


async def generate_image_async(workflow: dict) -> list[bytes]:
    client_id = str(uuid.uuid4())

    async with aiohttp.ClientSession() as session:
        # ジョブ投入
        async with session.post(
            f"{COMFYUI_URL}/prompt",
            json={"prompt": workflow, "client_id": client_id}
        ) as r:
            data = await r.json()
            prompt_id = data["prompt_id"]

        # WebSocket で完了まで待機
        async with session.ws_connect(
            f"ws://127.0.0.1:8188/ws?clientId={client_id}"
        ) as ws:
            async for msg in ws:
                if msg.type == aiohttp.WSMsgType.TEXT:
                    payload = json.loads(msg.data)
                    if payload["type"] == "executing":
                        if (payload["data"].get("node") is None
                                and payload["data"].get("prompt_id") == prompt_id):
                            break  # 完了

        # 画像取得
        async with session.get(
            f"{COMFYUI_URL}/history/{prompt_id}"
        ) as r:
            history = await r.json()

        images = []
        for node_output in history[prompt_id]["outputs"].values():
            for img_info in node_output.get("images", []):
                async with session.get(
                    f"{COMFYUI_URL}/view", params={
                        "filename": img_info["filename"],
                        "subfolder": img_info["subfolder"],
                        "type": img_info["type"],
                    }
                ) as r:
                    images.append(await r.read())
        return images


if __name__ == "__main__":
    import sys
    # 使い方: python3 async_client.py workflow_api.json
    with open(sys.argv[1]) as f:
        workflow = json.load(f)
    image_bytes = asyncio.run(generate_image_async(workflow))
    for i, b in enumerate(image_bytes):
        with open(f"output_{i:03d}.png", "wb") as f:
            f.write(b)
    print(f"{len(image_bytes)} image(s) saved.")

よくあるエラーと解決策

ComfyUI を外部から叩く場合の注意

デフォルトの ComfyUI は 127.0.0.1（localhost）でのみ待ち受けます。LAN 内の別マシンや Docker コンテナの外から接続するには --listen 0.0.0.0 オプションが必要です。

まとめ

ComfyUI の Python API で押さえておくべき点をまとめます。

• REST（/prompt POST）と WebSocket（/ws）の組み合わせが基本パターン
• client_id（UUID）が両者の紐付けキー。WebSocket を使うなら必ず渡す
• 完了検知は executing イベントで node == null になった瞬間
• 画像取得は /history/{prompt_id} → /view?filename=... の2ステップ
• 非同期が必要なら aiohttp + aiohttp.ClientSession.ws_connect() で置き換えられる

GUI を使わない自動化フローが整うと、複数プロンプトのバッチ処理やパラメータスイープが現実的な手間でできるようになります。モデルのロードコストはプロセスを起動したままにすることで吸収できるので、処理性能は思っているより高くなります。

VPS に ComfyUI を常駐させて外部から叩きたい場合は、RunPod で ComfyUI をクラウド GPU に展開する記事も参考にしてください。

関連記事（ComfyUIシリーズ）

• RunPodでComfyUIを使う完全ガイド — クラウドGPUで高速に画像生成（GPU不要）
• UbuntuにComfyUIを構築する方法 — 自分のUbuntu機にローカル導入する手順
• ComfyUI ワークフロー完全ガイド — ノードベース操作を基礎から実践まで
• ComfyUI カスタムノード厳選ガイド — ComfyUI-Managerと必須拡張の導入