AI活用

ComfyUIをPythonから実行する|公式SDKで生成画像を保存する方法

ComfyUIをPythonから動かすには、公式SDK「comfy-sdk」(Comfy API v2用、Beta 0.1.x)を使い、ローカルはcomfy-api-proxy経由で接続します。venvでの導入と接続確認、PromptとSeedの引数化、完了待ちと保存、複数実行のエラー処理、client層とmain処理の分離を解説します。

この記事の目次
  1. 結論:接続先を先に固定し、PromptとSeedを引数にした関数を作る
  2. Python連携の構成をどう決めるか
  3. SDKの現行状態(2026年8月22日確認)
  4. SDKをどう導入して接続確認するか
  5. Workflowにどう値を渡すか
  6. 生成完了をどう待って保存するか
  7. 再現実験:同じ関数を2回呼ぶ
  8. 複数実行のエラーをどう処理するか
  9. 再利用できる関数にどう分けるか
  10. よくある質問
  11. asyncioで使えますか?
  12. プロキシを使わずに8188へ直接つなげますか?
  13. Comfy Cloudに移すときコードはどれくらい変わりますか?
  14. まとめ

ComfyUIをPythonから動かすには、Comfy-Orgの公式SDK「comfy-sdk」を使い、ローカルのComfyUIにはcomfy-api-proxy経由で接続します。2026年8月22日時点でSDKはBeta(0.1.x)で、APIの形は変わり得ると公式が明記しています。接続先の固定から、引数化、保存、複数実行のエラー処理、client層の分離までを順に作ります。

情報確認日:2026年8月22日(日本時間)

結論:接続先を先に固定し、PromptとSeedを引数にした関数を作る

責任範囲

  • 公式Python SDKの現行状態と、ローカル/Cloudの接続先の決め方を示す
  • venvにSDKを入れ、プロキシ経由で接続確認する手順を固定する
  • Prompt・Seed・入力画像を引数にしてWorkflowに渡し、完了を待って画像を保存する
  • 複数実行で失敗したjobを無限に再試行しない設計と、main処理とclient層の分離を行う
  • API経路の全体像とServer APIの直接呼び出しはComfyUI APIの使い方に任せる
スポンサーリンク

Python連携の構成をどう決めるか

最初に決めるのは「どこで動いているComfyUIにつなぐか」です。接続先は次の三つのどれかで、既定はComfy Cloudです。何も設定せずに実行するとCloudに向きます。

  • ローカル(セルフホスト)COMFY_BASE_URL=http://127.0.0.1:8189api_keyは渡さない。comfy-api-proxyがComfyUI(8188)の前で動いている前提
  • Comfy CloudCOMFY_BASE_URL未設定(既定)、api_key必須
  • serverless deploymenthttps://<deployment>.run.comfy.appapi_key必須

この記事はローカル前提です。SDKはプロキシ(8189)と話すので、Server APIの8188を指定しても動きません。

SDKの現行状態(2026年8月22日確認)

PyPIにcomfy-sdkが存在し、最新は0.1.8(2026年8月13日公開)、リポジトリはComfy-Org配下のcomfy-python-sdkです。要件はPython 3.10以上。公式は「Beta。0.1.xの間はAPIの形が変わり得る」としています。

SDKをどう導入して接続確認するか

venvでSDKの版を固定し、プロキシを別ターミナルで常駐させます。接続確認はWorkflowを1回流すだけです。

# ComfyUI(127.0.0.1:8188)は起動済みの前提
python3 -m venv .venv
source .venv/bin/activate          # Windows は .venv\Scripts\activate
pip install comfy-sdk==0.1.8       # 確認日時点の版に固定
pip install comfy-api-proxy

# 別ターミナルで常駐させる(既定で 127.0.0.1:8189)
comfy-api-proxy

# SDKの接続先をローカルプロキシにする
export COMFY_BASE_URL="http://127.0.0.1:8189"
# hello.py — 接続確認。workflow_api.json は File → Export Workflow (API) で書き出したもの
from comfy_sdk import Comfy

client = Comfy()  # COMFY_BASE_URL を読む。セルフホストでは api_key を渡さない
wf = client.workflows.from_file("workflow_api.json")
job = client.run(wf)  # 送信して完了までポーリング
print(job.status, len(job.outputs))

succeededと出力数が出れば、プロキシ・ComfyUI・JSONが噛み合っています。Comfy(api_key=...)はキーワード引数専用で、位置引数でURLを渡す古い書き方はTypeErrorになります。

Workflowにどう値を渡すか

Workflow JSONは固定し、変わる値はwf.set_input(node_id, input_name, value)で差し込みます。node_idはJSONのキー、input_nameはそのノードのinputsのキーです。入力画像はclient.assets.from_file()のハンドルをそのまま値にします。送信時にハッシュされ、サーバーに同じ内容があればアップロードは省かれます。

# generate.py
from pathlib import Path
from comfy_sdk import Comfy

# 自分の workflow_api.json に合わせて変えるのはここだけ
NODE_POSITIVE = "6"   # CLIPTextEncode(正のPrompt)
NODE_SAMPLER = "3"    # KSampler
NODE_LOAD_IMAGE = "10"  # LoadImage
NODE_SAVE = "9"       # SaveImage

def build_workflow(client: Comfy, prompt: str, seed: int, image: Path | None = None):
    wf = client.workflows.from_file("workflow_api.json")
    wf.set_input(NODE_POSITIVE, "text", prompt)
    wf.set_input(NODE_SAMPLER, "seed", seed)
    if image is not None:
        wf.set_input(NODE_LOAD_IMAGE, "image", client.assets.from_file(str(image)))
    return wf

入力名に迷ったら、JSONでclass_typeKSamplerのノードのinputsを見ます。ノードキーは書き出し直すと変わり得るので定数にまとめます。入力画像をServer APIで直接アップロードする方法は別記事で扱います。

生成完了をどう待って保存するか

client.run(wf)submit()result()をまとめたもので、完了までポーリングします。途中経過はsubmit()後のjob.events()(SSE)で読めますが、READMEは「再接続時に見逃した分は再送されないため、完了判定はポーリングが正」としています。保存の判断にはevents()ではなくresult()run()の戻り値)を使い、job.get_outputs(NODE_SAVE)を回してout.to_file(path)で書き出します。

各出力はto_fileto_bytesget_download_urljob_idを持ちます。seedやpromptは含まれないので、ファイル名かログに自分で残します。

再現実験:同じ関数を2回呼ぶ

Seedを固定した関数を2回呼び、同じ画像が出るかを確かめます。手順と記録表だけを示します。

  1. GUIで動くWorkflowをFile → Export Workflow (API)で書き出す
  2. 最後の節のgenerate("a quiet harbor at dawn, watercolor", 12345, Path("out"))を2回呼ぶ
  3. 保存された2枚のsha256をshasum -a 256 out/*.pngで比べる
  4. seedだけ変えて3回目を呼び、ハッシュが変わることも確かめる
prompt seed job.status 保存ファイル sha256(先頭8桁) 所要時間
1
2(同条件)
3(seed変更)

1回目と2回目のハッシュが一致すれば、SDK経由でも決定論的に生成できています。一致しないときはseed以外に毎回変わる入力がないかを疑い、3回目で変わらないならノードキーがKSamplerを指していない可能性が高いです。

複数実行のエラーをどう処理するか

Promptのリストをまとめて流すと、1件の失敗が全体を止めるか、再試行し続けて止まらなくなるかに転びがちです。再試行してよい失敗とそうでない失敗を分け、回数と期限で必ず止めます。

例外 意味 再試行
JobFailed 実行中にノードが失敗(e.errorにnode情報) しない。同じ入力は同じ理由で失敗する
QueueFull Queueが満杯。submit()は自動で再試行する 自動再試行の後も出るなら間隔を空けて回数制限付きで
接続エラー(httpx由来) プロキシかComfyUIに届かない 回数制限付きで。起動を確認する
# batch.py
import time
from comfy_sdk import Comfy, JobFailed, QueueFull
from generate import build_workflow, NODE_SAVE

MAX_RETRY = 2        # 再試行してよい失敗だけに適用
DEADLINE_SEC = 600   # バッチ全体の期限

def run_batch(items, out_dir):
    client, report, started = Comfy(), {}, time.monotonic()
    for prompt, seed in items:
        if time.monotonic() - started > DEADLINE_SEC:
            report[seed] = "skipped: deadline"
            continue
        for attempt in range(MAX_RETRY + 1):
            try:
                job = client.run(build_workflow(client, prompt, seed))
                for i, out in enumerate(job.get_outputs(NODE_SAVE)):
                    out.to_file(str(out_dir / f"{seed}-{i}.png"))
                report[seed] = "ok"
                break
            except JobFailed as e:
                report[seed] = f"failed: {e.error}"  # 再試行しない
                break
            except QueueFull:
                report[seed] = "failed: queue full"
                time.sleep(5 * (attempt + 1))
    return report

失敗したjobを再試行しないのは、同じ入力なら同じノードが同じ理由で失敗するからです。reportのnode情報からGUIで該当ノードを確かめます。Queue制御は別記事で扱います。

再利用できる関数にどう分けるか

SDKがBetaである以上、メソッド名が変わる前提で設計します。comfy_sdkimportするファイルを一つに限り、main処理はそのファイルが返す自分の型だけを扱います。

# comfy_client.py — comfy_sdk を import するのはこのファイルだけ
from dataclasses import dataclass, field
from pathlib import Path
from comfy_sdk import Comfy, JobFailed

@dataclass
class GenerateResult:
    ok: bool
    files: list[Path] = field(default_factory=list)
    reason: str = ""

def generate(prompt: str, seed: int, out_dir: Path) -> GenerateResult:
    try:
        client = Comfy()
        wf = client.workflows.from_file("workflow_api.json")
        wf.set_input("6", "text", prompt)
        wf.set_input("3", "seed", seed)
        job = client.run(wf)
        files = []
        for i, out in enumerate(job.get_outputs("9")):
            path = out_dir / f"{seed}-{i}.png"
            out.to_file(str(path))
            files.append(path)
        return GenerateResult(ok=True, files=files)
    except JobFailed as e:
        return GenerateResult(ok=False, reason=str(e.error))

# main.py — SDKの名前を知らない
from comfy_client import generate
print(generate("a quiet harbor at dawn, watercolor", 12345, Path("out")))

SDKを上げてメソッド名が変わっても直すのはcomfy_client.pyだけで、main.pyGenerateResultしか知りません。版はrequirements.txtで固定し、ノードキーは設定に出し、確認日とSDKの版をREADMEに残します。

よくある質問

asyncioで使えますか?

使えます。AsyncComfyがあり、async with AsyncComfy() as client:の形で同じメソッドをawaitで呼びます。

プロキシを使わずに8188へ直接つなげますか?

SDKでは無理です。Server API(8188)はAPI v2の形を持ちません。プロキシを増やしたくないなら、urllibでServer APIを直接呼びます。最小例はComfyUI APIの使い方にあります。

Comfy Cloudに移すときコードはどれくらい変わりますか?

COMFY_BASE_URLを外し、Comfy(api_key="comfyui-...")でキーを渡す差分です。課金や保存期間は移す前に公式で確認してください。

まとめ

PythonからComfyUIを動かす公式SDKはcomfy-sdkとして実在し、ローカルではcomfy-api-proxyを挟み、COMFY_BASE_URLで接続先を固定してから使います。Workflow JSONは固定し、PromptとSeedはset_inputで差し込み、run()の完了後にget_outputsから保存します。

複数実行ではJobFailedを再試行せず、回数と期限で必ず止めてください。SDKはBetaなので、版を固定し、comfy_sdkを触るファイルを一つに限る分離が更新への一番安い保険です。

スポンサーリンク