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:8189、api_keyは渡さない。comfy-api-proxyがComfyUI(8188)の前で動いている前提 - Comfy Cloud:
COMFY_BASE_URL未設定(既定)、api_key必須 - serverless deployment:
https://<deployment>.run.comfy.app、api_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_typeがKSamplerのノードの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_file・to_bytes・get_download_urlとjob_idを持ちます。seedやpromptは含まれないので、ファイル名かログに自分で残します。
再現実験:同じ関数を2回呼ぶ
Seedを固定した関数を2回呼び、同じ画像が出るかを確かめます。手順と記録表だけを示します。
- GUIで動くWorkflowを
File → Export Workflow (API)で書き出す - 最後の節の
generate("a quiet harbor at dawn, watercolor", 12345, Path("out"))を2回呼ぶ - 保存された2枚のsha256を
shasum -a 256 out/*.pngで比べる - 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_sdkをimportするファイルを一つに限り、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.pyはGenerateResultしか知りません。版は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を触るファイルを一つに限る分離が更新への一番安い保険です。