Metadata-Version: 2.4
Name: comfyui-async-api-tool
Version: 0.1.0
Summary: Async Python client for ComfyUI: dispatch, stream progress, fetch images.
Author: ja1496
License: MIT
Project-URL: Homepage, https://github.com/ja1496/comfyui-async-api-tool
Project-URL: Repository, https://github.com/ja1496/comfyui-async-api-tool
Keywords: comfyui,async,aiohttp,stable-diffusion,workflow
Classifier: Development Status :: 3 - Alpha
Classifier: Intended Audience :: Developers
Classifier: License :: OSI Approved :: MIT License
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.10
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Topic :: Multimedia :: Graphics
Requires-Python: >=3.10
Description-Content-Type: text/markdown
License-File: LICENSE
Requires-Dist: aiohttp>=3.9
Provides-Extra: aiofiles
Requires-Dist: aiofiles>=23.2; extra == "aiofiles"
Provides-Extra: dev
Requires-Dist: aiofiles>=23.2; extra == "dev"
Requires-Dist: pytest>=8; extra == "dev"
Requires-Dist: pytest-asyncio>=0.23; extra == "dev"
Dynamic: license-file

# comfyui-async-api-tool

An async Python client for [ComfyUI](https://github.com/comfyanonymous/ComfyUI).
Pick a server, submit a workflow, stream progress, get the image — that's it.

一个 [ComfyUI](https://github.com/comfyanonymous/ComfyUI) 的异步 Python 客户端。
选机、提交、流式进度、取图——就这些。

[English](#english) · [中文](#中文)

---

## 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.

```python
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

```bash
pip install comfyui-async-api-tool
```

Only hard dependency is `aiohttp`. Python 3.10+.

### Features

- **Least-loaded dispatch** across multiple ComfyUI servers
- **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

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

The least busy server is chosen per run (remote queue + locally dispatched
tasks). A node that was recently online still gets queued if a probe fails
while it is busy rendering.

### Upload an image

```python
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()`

```python
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:

```python
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:

```python
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:

```python
# 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

```python
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:

```python
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

```bash
pip install -e .[dev]
pytest
```

---

## 中文

### 定位

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

```python
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())
```

### 安装

```bash
pip install comfyui-async-api-tool
```

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

### 特性

- **多机负载均衡**：自动挑选负载最低的 ComfyUI 服务器
- **实时进度**：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 -> {字段: 值}`，提交前会直接写入工作流。

### 多服务器

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

每次运行自动选择负载最低的机器（远程队列 + 本地已派发任务）。
若某台机器正在出图导致探测失败，只要它最近在线过，仍会排队而非报错。

### 上传图片

```python
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()`

```python
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、每次换不同的种子：

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

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

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

```python
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()` 会自动处理随机种子、默认值与类型格式化：

```python
# 从自动生成的配置开始：默认全部不暴露，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 配置骨架 |

### 探测服务器队列

```python
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*` 等字段默认按浮点处理。可以按实例自行扩充，
且不会改动内置默认值：

```python
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")`。

### 开发

```bash
pip install -e .[dev]
pytest
```

---

## License

MIT
