comfyui-async-api-tool

An async Python client for ComfyUI. Pick a server, submit a workflow, stream progress, get the image — that's it.

一个 ComfyUI 的异步 Python 客户端。选机、提交、流式进度、取图——就这些。

English

Positioning

Built for service backends (FastAPI, aiohttp apps, workers): you own the ClientSession, get a streamed event loop (start → progress → image), optionally dispatch across several ComfyUI machines, and edit workflows by node id / field. It is an orchestration kernel, not a GUI wrapper or a one-shot script helper.

import asyncio
import aiohttp
from comfyui_async_api_tool import ComfyClient

async def main():
    async with aiohttp.ClientSession() as session:
        client = ComfyClient(session, servers=["http://127.0.0.1:8188"])
        async for event in client.run(workflow, values={"3": {"seed": 42}}):
            if event["type"] == "progress":
                print(f"\r{event['data']:.0f}%", end="")
            elif event["type"] == "image":
                open("out.png", "wb").write(event["data"])

asyncio.run(main())

Install

pip install comfyui-async-api-tool

Only hard dependency is aiohttp. Python 3.10+.

Features

Least-loaded dispatch across multiple ComfyUI servers
- Probe every host concurrently; one unresponsive machine is skipped (no hang)
- Equal load → process-wide round-robin (not always the first URL)
- Selection uses a short single probe (retries=0); all-offline falls back to recently-online nodes
Live progress over WebSocket, smoothed, with a queue-polling fallback
Result fetch from /history + /view, with execution timing
Image upload from bytes, data-URI, URL, or local path
Workflow helpers to inspect nodes and apply user values

Events

run() is an async generator of events:

event	data
`start`	`{"client_id", "prompt_id", "server_url"}`
`progress`	`0.0` – `100.0`
`image`	raw `bytes` (+ `generation_ms` when available)
`error`	`msg` describing what went wrong

values maps node_id -> {field: value} and is written straight into the workflow before submission.

Multiple servers

client = ComfyClient(session, servers=[
    "http://gpu-1:8188",
    "http://gpu-2:8188",
])

Each run picks the least busy server (remote queue + process-wide locally dispatched tasks). Hosts are probed in parallel: a dead/slow host is skipped so the others still compete. When several machines share the same load, picks rotate round-robin across process lifetime (safe even if you construct a new ComfyClient per request). If every probe fails while nodes are busy rendering, recently-online hosts remain eligible for degraded dispatch.

Upload an image

resp = await client.upload_image("http://127.0.0.1:8188", "input.png")
# -> {"name": "....png", "subfolder": "", "type": "input"}

Accepts bytes, a data:image/...;base64,... URI, an http(s):// URL, or a local file path.

Inspect & edit a workflow

WorkflowJson and ComfyClient.run() are designed to work together: the first one tells you what can be changed, the second one applies the changes and runs. There are three ways to combine them.

Way 1 (recommended): inspect, then pass `values` to `run()`

import json
from comfyui_async_api_tool import ComfyClient, WorkflowJson

workflow = json.load(open("workflow.json", encoding="utf-8"))

wj = WorkflowJson()

# 1. See which nodes exist and what they do
for n in wj.list_all_nodes(workflow):
    print(n["node_id"], n["class_type"], n["title"])

# 2. List every writable scalar field (skips link references automatically)
#    -> find the node_id / field_key you want to override
for f in wj.extract_scalar_inputs(workflow):
    print(f["node_id"], f["field_key"], f["value_type"], "=", f["current_value"])

# 3. Pass overrides as {node_id: {field_key: value}}; run() writes them into
#    a private copy of the workflow before submitting
async for event in client.run(workflow, values={
    "3": {"seed": 42, "steps": 28},                  # KSampler
    "6": {"text": "a cat sitting on the moon"},      # positive prompt
    "7": {"text": "blurry, low quality"},            # negative prompt
    "5": {"width": 832, "height": 1216},             # latent size
}):
    ...

run() never mutates your workflow dict — it deep-copies before applying values, so you can reuse the same dict in a loop with different seeds:

for seed in range(4):
    async for event in client.run(workflow, values={"3": {"seed": seed}}):
        ...  # each run starts from the pristine workflow

Way 2: edit first with `batch_modify`, then run

Useful when you want to review or reuse the final JSON yourself:

edited = wj.batch_modify(workflow, [
    {"top_key": "3", "target_key": "seed", "new_value": 123},
    {"top_key": "6", "target_key": "text", "new_value": "a dog"},
])
# `edited` is the same dict object (edited in place) — copy first if needed:
# import copy; edited = wj.batch_modify(copy.deepcopy(workflow), [...])

async for event in client.run(edited):   # no `values` needed anymore
    ...

Way 3 (advanced): `exposed_fields` for app-style parameter panels

If you are building a UI where users tweak a fixed set of parameters, describe them once as exposed_fields; run() then handles seeds, defaults and type formatting automatically:

# Start from an auto-generated config: nothing exposed, seeds auto-randomized
fields = wj.default_exposed_fields(wj.extract_scalar_inputs(workflow))

# Mark what your users may edit
for f in fields:
    if f["node_id"] == "6" and f["field_key"] == "text":
        f["expose"] = True
    if f["node_id"] == "5" and f["field_key"] in ("width", "height"):
        f["expose"] = True
# fields flagged auto_seed=True get a fresh random seed on every run

async for event in client.run(workflow, exposed_fields=fields, values={
    "6": {"text": "a cat"},
    "5": {"width": 768},
}):
    ...

Handy helpers used above:

method	purpose
`list_all_nodes(workflow)`	every node's id, title, class_type, inputs
`extract_scalar_inputs(workflow)`	all writable scalar fields
`find_nodes_with_input_key(workflow, "seed")`	locate nodes by field name
`find_last_output_image_node(workflow)`	the SaveImage/PreviewImage node
`batch_modify(workflow, mods)`	apply edits in place
`default_exposed_fields(fields)`	scaffold an exposed_fields config

Probe a server's queue

from comfyui_async_api_tool import queue

snap = await queue.probe_queue(session, "http://127.0.0.1:8188")
# snap.online, snap.running, snap.pending, snap.queue_remaining

Custom float params

Fields like cfg / denoise / *strength* are treated as floats out of the box. Add your own per instance — the built-in defaults are never mutated:

wj = WorkflowJson(extra_float_keys={"guidance", "eta"})
client = ComfyClient(session, servers=[...], extra_float_keys={"guidance"})

Notes

The aiohttp.ClientSession is owned by you; the client never closes it.
run() expects the workflow already loaded as a dict, and never mutates it.
To read a workflow file asynchronously, install the extra pip install comfyui-async-api-tool[aiofiles], then await WorkflowJson().load_json_async("workflow.json").

Development

pip install -e .[dev]
pytest

中文

定位

面向 服务端编排（FastAPI、aiohttp 应用、后台 worker）： session 由你持有，任务以事件流推进（start → progress → image），可在多台 ComfyUI 间按负载选机，并按节点 id / 字段改工作流。它是编排内核，不是 GUI 封装，也不是一次性脚本工具。

import asyncio
import aiohttp
from comfyui_async_api_tool import ComfyClient

async def main():
    async with aiohttp.ClientSession() as session:
        client = ComfyClient(session, servers=["http://127.0.0.1:8188"])
        async for event in client.run(workflow, values={"3": {"seed": 42}}):
            if event["type"] == "progress":
                print(f"\r{event['data']:.0f}%", end="")
            elif event["type"] == "image":
                open("out.png", "wb").write(event["data"])

asyncio.run(main())

安装

pip install comfyui-async-api-tool

唯一硬依赖是 aiohttp，要求 Python 3.10+。

特性

多机负载均衡：自动挑选负载最低的 ComfyUI 服务器
- 并发探测全部地址；单台无响应只跳过，不拖死整次选机
- 负载并列时进程级轮询（即使每次新建 ComfyClient 也不会总打第一台）
- 选机短探测（retries=0）；全部假离线时降级到「最近在线」节点
实时进度：WebSocket 推送 + 平滑处理，并以队列轮询兜底
结果获取：从 /history + /view 取图，附带执行耗时
图片上传：支持 bytes、data-URI、网络 URL、本地路径
工作流工具：节点检查、字段提取、批量改值

事件

run() 是一个异步生成器，产出以下事件：

事件	data 内容
`start`	`{"client_id", "prompt_id", "server_url"}`
`progress`	`0.0` – `100.0`
`image`	图片原始 `bytes`（可能附带 `generation_ms`）
`error`	`msg` 错误描述

values 的结构为 节点id -> {字段: 值}，提交前会直接写入工作流。

多服务器

client = ComfyClient(session, servers=[
    "http://gpu-1:8188",
    "http://gpu-2:8188",
])

每次运行自动选择负载最低的机器（远程队列 + 进程级本地已派发任务）。全部地址并发探测：有一台死机/超时只跳过，其余照常参与比较；负载相同时在进程内轮询分配，避免总打列表第一台。若出图导致探测全失败，只要节点在「最近在线」窗口内，仍会降级排队而非直接报错。

上传图片

resp = await client.upload_image("http://127.0.0.1:8188", "input.png")
# -> {"name": "....png", "subfolder": "", "type": "input"}

支持 bytes、data:image/...;base64,...、http(s):// URL 或本地文件路径。

检查与修改工作流

WorkflowJson 与 ComfyClient.run() 是配套设计的：前者告诉你哪里能改，后者负责改完并跑起来。共有三种联用方式。

方式一（推荐）：先检查，再把 `values` 传给 `run()`

import json
from comfyui_async_api_tool import ComfyClient, WorkflowJson

workflow = json.load(open("workflow.json", encoding="utf-8"))

wj = WorkflowJson()

# 1. 查看有哪些节点、各自的作用
for n in wj.list_all_nodes(workflow):
    print(n["node_id"], n["class_type"], n["title"])

# 2. 列出所有可写标量字段（自动跳过连线引用）
#    从中找到你想覆盖的 node_id / field_key
for f in wj.extract_scalar_inputs(workflow):
    print(f["node_id"], f["field_key"], f["value_type"], "=", f["current_value"])

# 3. 按 {节点id: {字段: 值}} 传参；run() 会先把这些值写入
#    workflow 的私有副本，再提交执行
async for event in client.run(workflow, values={
    "3": {"seed": 42, "steps": 28},                  # KSampler 采样器
    "6": {"text": "a cat sitting on the moon"},      # 正向提示词
    "7": {"text": "blurry, low quality"},            # 负向提示词
    "5": {"width": 832, "height": 1216},             #  latent 尺寸
}):
    ...

run() 不会改动你传入的 workflow——写入前会先深拷贝，因此可以在循环里复用同一个 dict、每次换不同的种子：

for seed in range(4):
    async for event in client.run(workflow, values={"3": {"seed": seed}}):
        ...  # 每次都从原始工作流出发，互不污染

方式二：先用 `batch_modify` 改好，再运行

适合想自己检查或复用最终 JSON 的场景：

edited = wj.batch_modify(workflow, [
    {"top_key": "3", "target_key": "seed", "new_value": 123},
    {"top_key": "6", "target_key": "text", "new_value": "a dog"},
])
# 注意：batch_modify 是原地修改，edited 与 workflow 是同一个对象。
# 如需保留原对象，先拷贝：
# import copy; edited = wj.batch_modify(copy.deepcopy(workflow), [...])

async for event in client.run(edited):   # 此时无需再传 values
    ...

方式三（进阶）：`exposed_fields` 做应用式参数面板

如果你在做一个让用户调节固定参数的界面，可以把参数描述成 exposed_fields；run() 会自动处理随机种子、默认值与类型格式化：

# 从自动生成的配置开始：默认全部不暴露，seed 类字段自动随机
fields = wj.default_exposed_fields(wj.extract_scalar_inputs(workflow))

# 标记允许用户编辑的字段
for f in fields:
    if f["node_id"] == "6" and f["field_key"] == "text":
        f["expose"] = True
    if f["node_id"] == "5" and f["field_key"] in ("width", "height"):
        f["expose"] = True
# auto_seed=True 的字段每次运行都会写入新的随机种子

async for event in client.run(workflow, exposed_fields=fields, values={
    "6": {"text": "a cat"},
    "5": {"width": 768},
}):
    ...

上面用到的常用方法：

方法	作用
`list_all_nodes(workflow)`	列出每个节点的 id、标题、class_type、inputs
`extract_scalar_inputs(workflow)`	提取所有可写标量字段
`find_nodes_with_input_key(workflow, "seed")`	按字段名定位节点
`find_last_output_image_node(workflow)`	找到 SaveImage/PreviewImage 出图节点
`batch_modify(workflow, mods)`	原地批量修改
`default_exposed_fields(fields)`	生成 exposed_fields 配置骨架

探测服务器队列

from comfyui_async_api_tool import queue

snap = await queue.probe_queue(session, "http://127.0.0.1:8188")
# snap.online, snap.running, snap.pending, snap.queue_remaining

自定义浮点参数

cfg / denoise / *strength* 等字段默认按浮点处理。可以按实例自行扩充，且不会改动内置默认值：

wj = WorkflowJson(extra_float_keys={"guidance", "eta"})
client = ComfyClient(session, servers=[...], extra_float_keys={"guidance"})

说明

aiohttp.ClientSession 由你持有，客户端不会关闭它。
run() 要求工作流已加载为 dict，且不会改动它。
如需异步读取工作流文件，安装可选项 pip install comfyui-async-api-tool[aiofiles]，然后 await WorkflowJson().load_json_async("workflow.json")。

开发

pip install -e .[dev]
pytest

License

MIT

Name		Name	Last commit message	Last commit date
comfyui_async_api_tool		comfyui_async_api_tool
examples		examples
tests		tests
.gitignore		.gitignore
LICENSE		LICENSE
README.md		README.md
pyproject.toml		pyproject.toml

Latest commit

History

Folders and files

Repository files navigation

comfyui-async-api-tool

English

Positioning

Install

Features

Events

Multiple servers

Upload an image

Inspect & edit a workflow

Way 1 (recommended): inspect, then pass values to run()

Way 2: edit first with batch_modify, then run

Way 3 (advanced): exposed_fields for app-style parameter panels

Probe a server's queue

Custom float params

Notes

Development

中文

定位

安装

特性

事件

多服务器

上传图片

检查与修改工作流

方式一（推荐）：先检查，再把 values 传给 run()

方式二：先用 batch_modify 改好，再运行

方式三（进阶）：exposed_fields 做应用式参数面板

探测服务器队列

自定义浮点参数

说明

开发

License

About

Resources

Stars

Watchers

Forks

Releases

Packages

Contributors

Languages