# ComfyUI 工作流「直连提交」方法 —— 给其他 Agent 的实操说明

> 适用场景：不开浏览器、不进 ComfyUI 网页界面，直接用 HTTP 把任务提交给本机（或局域网）的 ComfyUI 引擎跑图/跑视频。
> 本文来自实战（MiniMax H3 视频工作流，约 15 段成片），所有代码都是实际跑通过的写法。

---

## 0. 一句话原理

ComfyUI 的网页界面点击「生成」时，本质是浏览器把一份 **API 格式的 JSON** POST 到 `http://<host>:8188/prompt`。
所以只要我们能造出同样的 JSON，就能绕过界面直接提交。关键在于：**界面用的 workflow JSON 和 API 用的 JSON 不是同一种格式**。

| | 界面格式（workflow.json） | API 格式 |
|---|---|---|
| 节点结构 | `{"nodes":[{id, type, widgets_values, inputs, outputs}], "links":[...]}` | `{"<节点id>": {"class_type":"...", "inputs":{...}}}` |
| 连线 | 独立 `links` 数组 | 内联成 `["源节点id", 输出槽位号]` |
| 参数 | `widgets_values` 数组，位置对应 | `inputs` 字典，按名字 |
| 用途 | 拖进网页界面 | 提交给 `/prompt` |

**核心难点：怎么把界面格式转成 API 格式。** 见第 2 节。

---

## 1. 判断引擎是否在线

```python
import urllib.request, json

HOST = "http://127.0.0.1:8188"

def alive():
    try:
        json.load(urllib.request.urlopen(f"{HOST}/system_stats", timeout=5))
        return True
    except Exception as e:
        print("引擎未启动:", e)
        return False
```

常用只读接口（都不需要鉴权）：

| 接口 | 用途 |
|---|---|
| `GET /system_stats` | 引擎存活 + 显存/设备信息 |
| `GET /object_info` | 全部节点类型定义（节点有哪些参数名、可选值） |
| `GET /queue` | 当前在跑 / 排队的任务 |
| `GET /history/<prompt_id>` | 某个任务的执行结果与输出文件名 |

**注意**：`GET /object_info` 返回体很大（几 MB），除非要查参数名，平时别拉。

---

## 2. 拿到 API 格式的 JSON（三种路，按推荐度排）

### 路 A：界面里的「导出 (API)」★最推荐

1. 网页界面打开工作流
2. 右上角菜单点 **Save (API Format)** / 「导出 (API)」
   - 新版前端：设置里先勾选 `Enable Dev mode Options`，顶部才会出现 `Save (API Format)` 按钮
3. 得到一个 `xxx_api.json`，就是可以直接 POST 的东西

这条路 100% 准确，**优先用它**。

### 路 B：从 `/history` 反推

跑过一次的任务，`GET /history/<prompt_id>` 返回体里的 `prompt` 字段就是当初提交的 API JSON。

```python
h = json.load(urllib.request.urlopen(f"{HOST}/history/{pid}", timeout=10))
api_json = h[pid]["prompt"][2]   # 注意：是下标 2
```
适合「别人跑过、我想要他的参数」的情况。

### 路 C：自己转（实在没有导出功能时）

从界面格式手转，规则：

```python
# 界面格式 nodes 里，每个节点：
#   n["id"]          -> API 的 key（转成字符串）
#   n["type"]        -> class_type
#   n["widgets_values"] + n["inputs"] 里非连线项 -> inputs
# 连线在 n["inputs"][i]["link"] 指向 links 里的 [link_id, 源节点, 源槽, 目标节点, 目标槽, 类型]
```

这条路容易错（尤其是 `widgets_values` 的位置对应、seed 类控件藏在数组里），**只在没有别的办法时走**，且必须先用一次极简工作流验证。

---

## 3. 提交的最小骨架

```python
import urllib.request, urllib.error, json, copy

HOST = "http://127.0.0.1:8188"

def submit(api_json, tag=""):
    data = json.dumps({"prompt": api_json}).encode()
    req = urllib.request.Request(
        f"{HOST}/prompt",
        data=data,
        headers={"Content-Type": "application/json"},
    )
    try:
        r = json.loads(urllib.request.urlopen(req, timeout=120).read())
        if r.get("node_errors"):
            print(f"  {tag} 节点错误:", json.dumps(r["node_errors"], ensure_ascii=False)[:500])
            return None
        pid = r.get("prompt_id", "")
        print(f"  {tag} OK  prompt_id={pid}")
        return pid
    except urllib.error.HTTPError as e:
        # 400 通常是 JSON 结构不合法 / 参数类型不对，body 里有详细原因
        print(f"  {tag} HTTP {e.code}: {e.read().decode(errors='ignore')[:500]}")
        return None
```

**成功返回**：`{"prompt_id": "...", "number": N, "node_errors": {}}`
**失败特征**：HTTP 400 / 500，或 `node_errors` 非空（这时代码里有价值的信息是每个节点缺了什么）。

---

## 4. 改参数的正确姿势

拿到 API JSON 后，它是一个字典，节点 id 是 key。改参数就是改对应节点的 `inputs`：

```python
base = json.load(open("xxx_api.json", encoding="utf-8"))

p = copy.deepcopy(base)                      # ★ 一定要深拷贝，多段循环复用模板时尤其重要

p["528"]["inputs"]["prompt"]        = "你的提示词"        # 文本提示词节点
p["521"]["inputs"]["filename_prefix"] = "output_tag"     # 输出文件名前缀
p["527"]["inputs"]["image"]         = "image_a.png"      # LoadImage 节点的图片名
p["529"]["inputs"]["value"]         = 18.0               # 时长（秒）
```

### 关键：怎么知道哪个节点 id 对应什么？

直接打印出来看，别猜：

```python
for nid, v in base.items():
    print(nid, v.get("class_type"), "->", {k: str(val)[:40] for k, val in v.get("inputs", {}).items()})
```

实跑例子（MiniMax H3 视频工作流，45 个节点里真正要动的只有 5~6 个）：

| 节点 id | class_type | 干什么 |
|---|---|---|
| 528 | （提示词节点） | 正文提示词 |
| 521 | VHS_VideoCombine | 输出文件名前缀 |
| 525/526/527/540/544 | LoadImage | 五个参考图槽位 |
| 529 | （时长控件） | 视频时长 |

---

## 5. 参考图必须先上传

`LoadImage` 的 `image` 字段填的是**文件名**，不是路径。文件必须已经存在于 ComfyUI 的 `input/` 目录下。

两种放法：

**法一（推荐）**：直接拷贝文件到 `<ComfyUI>/input/` 目录
```python
import shutil
shutil.copy2(r"F:\素材\my_face.png", r"<ComfyUI>\input\my_face.png")
```
缺点：文件名要自己管理，容易和别人的撞。

**法二**：走 `/upload/image` 接口
```python
import requests
requests.post(f"{HOST}/upload/image",
              files={"image": open("a.png", "rb")},
              data={"overwrite": "true"})
```
返回 `{"name": "...", "subfolder": ..., "type": "input"}`，`name` 就是要填进 `LoadImage` 的值。

**命名建议**：用 `前缀_语义_尺寸.png`（如 `头部_mask_1280.png`），多段循环时一眼能看出用的哪张。

---

## 6. 等出片 + 取结果

```python
import time

def wait_done(pid, timeout=1800, poll=10):
    t0 = time.time()
    while time.time() - t0 < timeout:
        h = json.load(urllib.request.urlopen(f"{HOST}/history/{pid}", timeout=15))
        if pid in h:
            entry = h[pid]
            status = entry.get("status", {})
            if status.get("completed") or status.get("status_str") == "success":
                return entry
            if status.get("status_str") == "error":
                print("执行失败:", json.dumps(status, ensure_ascii=False)[:800])
                return None
        time.sleep(poll)
    print("超时")
    return None

def outputs(entry):
    """取出输出文件列表（文件名 + 子目录 + 类型）"""
    res = []
    for nid, out in entry.get("outputs", {}).items():
        for key in ("images", "gifs", "videos", "audio"):
            for f in out.get(key, []) or []:
                res.append(f)   # {"filename":..., "subfolder":..., "type":"output"}
    return res
```

文件实际路径：`<ComfyUI>/output/<subfolder>/<filename>`

---

## 7. 批量跑多段：模板复用的坑

跑 10 段不同内容时，唯一的正确姿势是：

```python
base = json.load(open("tpl_api.json", encoding="utf-8"))

for tag, prompt, imgs, secs in PLAN:
    p = copy.deepcopy(base)              # ★ 每段都从模板深拷贝
    p["528"]["inputs"]["prompt"] = prompt
    p["521"]["inputs"]["filename_prefix"] = f"proj_{tag}"
    for nid, img in zip(["527","525","526","540","544"], imgs):
        p[nid]["inputs"]["image"] = img
    p["529"]["inputs"]["value"] = secs
    submit(p, tag)
    time.sleep(1)                        # 连续猛推容易触发队列竞争
```

**三个必踩的坑**：

1. **不深拷贝** → 上一段改的值带进下一段，最后全跑成同一段内容
2. **不留版本文件** → 出问题时无法复现。做法：每次提交前把 `p` 另存成 `_wf_api_<tag>.json`，留痕
3. **`filename_prefix` 不区分** → 输出全叠在一个前缀下，事后分不清哪段是哪段

---

## 8. 常见报错对照

| 现象 | 原因 | 处理 |
|---|---|---|
| `Connection refused` / `WinError 10061` | 引擎没起，或端口不是 8188 | 先 `system_stats` 探活；确认启动参数里的 `--port` |
| HTTP 400，body 里有 `required_input_missing` | 节点缺必填参数 | 对照 `/object_info` 里该 `class_type` 的 required 列表 |
| HTTP 400，`value_not_in_list` | 下拉框填了不存在的值（模型名、格式名写错） | 查 `/object_info` 该节点的可选值，别凭印象写 |
| `node_errors` 非空但任务已进队列 | 部分节点报错，其余照样执行 | 当失败处理，别等下一条 |
| 提交成功但一直没输出 | 卡在队列里 / 显存不够在死磕 | 查 `/queue`；查显存 |
| 输出文件名对但内容是上一段 | 没深拷贝 | 见上节坑 1 |

---

## 9. 一个可跑的最小示例

```python
"""最小验证：探活 -> 改提示词 -> 提交 -> 等完成 -> 打印输出"""
import urllib.request, urllib.error, json, copy, time

HOST = "http://127.0.0.1:8188"
TPL  = "workflow_api.json"      # 从界面导出的 API 格式

# 1) 探活
json.load(urllib.request.urlopen(f"{HOST}/system_stats", timeout=5))
print("引擎在线")

# 2) 载入模板 & 看清节点
base = json.load(open(TPL, encoding="utf-8"))
for nid, v in base.items():
    print(f"  {nid} {v.get('class_type')}")

# 3) 改参数（id 按上一步打印的实际值替换）
p = copy.deepcopy(base)
# p["528"]["inputs"]["prompt"] = "你的提示词"
# p["521"]["inputs"]["filename_prefix"] = "test_001"

# 4) 提交
data = json.dumps({"prompt": p}).encode()
req = urllib.request.Request(f"{HOST}/prompt", data=data,
                             headers={"Content-Type": "application/json"})
r = json.loads(urllib.request.urlopen(req, timeout=120).read())
pid = r["prompt_id"]
print("已提交", pid)

# 5) 等
while True:
    h = json.load(urllib.request.urlopen(f"{HOST}/history/{pid}", timeout=15))
    if pid in h and h[pid].get("status", {}).get("completed"):
        for nid, out in h[pid].get("outputs", {}).items():
            for k in ("images", "gifs", "videos"):
                for f in out.get(k, []) or []:
                    print("输出:", f.get("subfolder"), f.get("filename"))
        break
    time.sleep(10)
```

---

## 10. 给 Agent 的检查清单

提交前必查：

- [ ] 引擎 `system_stats` 通
- [ ] 用的是 **API 格式** JSON（不是界面格式）
- [ ] 每个 LoadImage 的图片**已在 input/ 里**，文件名和字段一致
- [ ] 提示词节点 id、输出前缀 id 确认过（打印 `class_type` 对过，不靠猜）
- [ ] 每段 `copy.deepcopy`
- [ ] `filename_prefix` 每段不同
- [ ] 提交前把该段 JSON 存盘留痕
- [ ] 参数（步数、分辨率、时长、seed）**一字不改地沿用用户界面里的值** —— 擅自改动会让结果无法与已有成片对比

提交后必查：

- [ ] 返回 `prompt_id` 且 `node_errors` 为空
- [ ] `/queue` 里能看到在跑
- [ ] 完成后核对输出文件实际大小（不是 0 字节、不是明显偏小）
- [ ] 需要画面判断的，抽帧看一眼再说「成功」

---

## 附：本文方法在哪些环境验证过

- ComfyUI 版本：2026.8.8 整合包
- 工作流：MiniMax H3 视频（单节点一体化，45 节点）
- 系统：Windows 10 + Python urllib（无第三方依赖）
- 成片量：约 15 段（10~18 秒/段），含 5 图参考、语音台词、数字人