Comfy SDK：ComfyUI 的 Python 与 TypeScript 客户端

用 Python 或 TypeScript 对 Comfy Cloud、自建部署或自有 ComfyUI 运行 ComfyUI 工作流，并处理资产、任务、实时事件和幂等重试。

[!NOTE] Comfy API v2 采用版本管理。v2 内的变更仅作增量扩展；不兼容的变更会在 v3 中发布。请通过反馈报告问题。

一个包，两种接口。 comfy-sdk（PyPI）和 @comfyorg/sdk（npm）提供了两个客户端，分别与两个不同的服务通信。方法名会在两者之间重复出现（run、submit、events），但含义各不相同，因此在复制代码片段之前，请先确认它构建的是哪个客户端。

两种接口互不封装，在你的 Comfy 工作区中创建一个 API 密钥即可同时用于两者。

[!NOTE] 两种语言对 Router 接口的暴露方式不同。在 Python 中，Comfy() 同时承载两者：client.models.run(...) 用于 Router，client.workflows / client.assets / client.jobs 用于 Cloud。在 TypeScript 中，它们是两个独立的导出，名称刻意相近：comfy（小写，模块级命名空间）包含 comfy.models，而 Comfy（类）是 Cloud 客户端，没有 .models。

本页介绍第二行：Comfy Cloud 和 Comfy API v2 客户端。如果您要使用模型路由器，请从 Comfy Router 快速入门开始。

Comfy SDK 让您的应用程序能够运行 ComfyUI 工作流并取回结果。您提交工作流，ComfyUI 执行它，然后您下载输出。同一份代码既可用于 Comfy Cloud，也可用于您自行托管的 ComfyUI 实例，只需更改基础 URL。任务标签和任务列表除外：任务标签只适用于 Comfy API 部署，Comfy Cloud 也暂不提供任务列表。

这些 SDK 是 Comfy API v2 的客户端。Comfy API v2 是一个版本化的 HTTP API，我们计划长期支持。未来发布的 ComfyUI 新版本不会破坏基于该 API 构建的集成。

人们通过这种方式构建的内容：

• 在其他应用程序（如 Blender 或 Krita）内部生成内容的插件
• 代表用户执行生成的面向消费者的应用
• 批处理流水线，例如对视频的每一帧运行同一个工作流
• 需要同时运行大量工作流的后端服务

[!NOTE] 这些 SDK 是从外部驱动 ComfyUI。如果您要编写在 ComfyUI 内部运行的自定义节点或前端扩展，请改用开发自定义节点。那是一套独立的 API。

安装

pip install comfy-sdk

npm i @comfyorg/sdk

Python 3.10 或更高版本。Node 22 或更高版本。

[!NOTE] 当前版本：0.5.0。 PyPI 上的 comfy-sdk 与 npm 上的 @comfyorg/sdk 一同发布，并共用同一个版本号。请安装最新版本（pip install comfy-sdk、npm install @comfyorg/sdk），或声明一个范围，而不是精确固定某个版本。两个生态对此的写法不同：comfy-sdk>=0.5 是一个下限，接受后续的次版本；而 @comfyorg/sdk@^0.5 是 npm 针对 0.5.x 的兼容范围，止步于 0.6.0 之前，因此当新的次版本发布时，插入符范围需要上调。如果你希望 npm 端也跟随后续的次版本，请使用 @comfyorg/sdk@>=0.5.0。如果本说明滞后，以注册表页面为准。

快速入门

上传输入图像，运行工作流，并将结果写入磁盘。

from comfy_sdk import Comfy

# Comfy Cloud
client = Comfy(api_key="comfyui-...")

wf = client.workflows.from_file("workflow_api.json")

asset = client.assets.from_file("photo.png")
wf.set_input("10", "image", asset)

job = client.run(wf)
for output in job.get_outputs("9"):
    output.to_file(output.name)

import { Comfy } from "@comfyorg/sdk";

// Comfy Cloud
const client = new Comfy({ apiKey: "comfyui-..." });

const wf = await client.workflows.fromFile("workflow_api.json");

const asset = client.assets.fromFile("photo.png");
wf.setInput("10", "image", asset);

const job = await client.run(wf);
await job.getOutputs("9")[0].toFile("out.png");

workflow_api.json 是以 API 格式 保存的工作流。"10" 和 "9" 是该文件中的节点 ID：即输入图像送入的节点，以及你希望获取结果的输出节点。

资产句柄是惰性的。photo.png 会在本地进行哈希计算，仅当服务器尚未拥有这些字节时才会被上传，因此使用相同的输入重新运行不会产生任何成本。

run() 提交任务并等待其达到终端状态。如要在其执行期间进行其他操作，请改用 submit()，并观察 事件流。

若要改为在你自己的 ComfyUI 上运行，请设置 COMFY_BASE_URL 并去掉密钥。请参阅下文。

选择基础 URL

基础 URL 来自 COMFY_BASE_URL 环境变量，而不是构造函数参数：

export COMFY_BASE_URL="https://<deployment>.run.comfy.app"  # Comfy API 部署
export COMFY_BASE_URL="http://127.0.0.1:8189"               # 自托管代理

每次构造客户端时都会读取该变量，它必须是 http(s) URL；未设置或留空即表示使用 Comfy Cloud。因此客户端本身在所有环境下都是相同的：

client = Comfy(api_key="comfyui-...")

const client = new Comfy({ apiKey: "comfyui-..." });

[!NOTE] 从早期版本升级？Comfy("<url>", "<key>") 现在是在设置 COMFY_BASE_URL 后使用 Comfy(api_key="<key>")。api_key 是仅限关键字参数，因此旧的位置参数调用会抛出 TypeError，而不会将 URL 静默当作 API 密钥读取。

Comfy Cloud

开箱即用。创建 API 密钥 并将其传递给客户端。

[!NOTE] API 访问需要付费的 Comfy Cloud 订阅。免费层级不包含此功能。同时可运行的任务数量取决于您的订阅层级。请参阅并发限制。

Comfy API 部署

通过 开发者平台 部署的工作流会有自己的端点。将 COMFY_BASE_URL 指向该端点并使用您的 API 密钥，与 Comfy Cloud 完全相同。本指南中的所有内容都以相同的方式工作，任务标签和任务列表除外，Comfy Cloud 暂不支持这两项。

Comfy API 部署可以使用其 Build 中包含的模型和自定义节点运行多个工作流。对于每个任务，get_workflow() 返回的工作流响应带有 format: "api"，并在其 .graph 属性中包含已执行的图。

您自己的 ComfyUI

对于自托管的 ComfyUI，v2 API 由 comfy-api-proxy 提供，这是一个与您的 ComfyUI 一起运行的小型开源服务。安装它、运行它，并设置 COMFY_BASE_URL="http://127.0.0.1:8189"：

pip install comfy-api-proxy
comfy-api-proxy
export COMFY_BASE_URL="http://127.0.0.1:8189"

关于配置和身份验证，请参阅自托管 ComfyUI 的 API 代理。

重试提交：幂等键

submit() 和 run() 在每次提交时都会发送一个 Idempotency-Key，而 Comfy API v2 对该键的契约是重复即拒绝，而非记录并重放。在把提交包进重试循环之前，请先阅读本节。

• 每次调用都会生成一个新的键。 用同一个工作流调用 submit() 两次，就是两次提交、两个任务和两笔计费。在 submit() 外面套一个天真的 for attempt in range(3)，得到的是双重扣费，而不是重试。
• 被复用的键会被拒绝，而不是被重放。 传入你自己的 idempotency_key（TypeScript 中为 idempotencyKey）可以让重试具备幂等性，而使用该键的第二次请求会以 422 idempotency_key_reuse 失败，而不是返回第一个任务。各 SDK 会抛出 IdempotencyKeyReuse。这与“幂等重试”通常的含义正好相反，因此需要显式处理。
• 恢复方式是去找回任务，而不是重新提交。 遇到 IdempotencyKeyReuse 时，第一次尝试很可能已经创建了任务。用 client.jobs.get(job_id) 取回它。该查找需要你已存储的 id，因此它只能恢复“提交已返回、且你在故障发生前持久化了 id”这一情形。如果连接在记录 id 之前就断开了，那就没有可查找的对象，也没有自动恢复的办法：下面的示例选择重新抛出异常交由人工处理，而不是在已占用的键下重新提交。
• 当提交确定失败时，该键会被释放。 验证错误、积分不足被拒绝或队列已满被拒绝都不会创建任务，并会释放该键，因此在该键下重新提交是没问题的。在结果不明的失败之后（读取超时、请求中途连接断开），该键仍保持占用：此时应查找任务，而不是重新提交。
• 键会在 24 小时后过期，且必须非空、由可打印 ASCII 字符组成并符合长度限制。无效的键会在发出任何请求之前抛出 ValueError，因此显式的 "" 绝不会静默回退到一个自动生成的键。

因此，一个可以安全运行多次的重试必须在各次尝试之间携带两样东西：键，让服务器能够区分重试与新的提交；以及任务ID，让你能够找到已被占用的键所创建的任务。

import uuid
from comfy_sdk import IdempotencyKeyReuse

# 这两个值都应该放在你自己的存储里，以该请求对你而言
# 的含义作为键。`key` 只在第一次尝试时生成一次；`job_id`
# 在提交返回后立即填入。
state = load_submission_state()  # {"key": ..., "job_id": ...} or {}
key = state.get("key") or str(uuid.uuid4())
save_submission_state(key=key)

try:
    job = client.submit(wf, idempotency_key=key)
    save_submission_state(key=key, job_id=job.id)
except IdempotencyKeyReuse:
    # 此前在该键下的一次尝试已经到达提交环节，因此可能已存在一个任务。
    # 不要重新提交：去找回它。
    job_id = state.get("job_id")
    if job_id is None:
        raise  # 没有记录可供查找；这需要人工处理
    job = client.jobs.get(job_id)

import { IdempotencyKeyReuse } from "@comfyorg/sdk";

// 这两个值都应该放在你自己的存储里，以该请求对你而言
// 的含义作为键。`key` 只在第一次尝试时生成一次；`jobId`
// 在提交返回后立即填入。
const state = await loadSubmissionState(); // { key?, jobId? }
const key = state.key ?? crypto.randomUUID();
await saveSubmissionState({ key });

let job;
try {
  job = await client.submit(wf, { idempotencyKey: key });
  await saveSubmissionState({ key, jobId: job.id });
} catch (err) {
  // 此前在该键下的一次尝试已经到达提交环节，因此可能已存在一个任务。
  // 不要重新提交：去找回它。
  if (!(err instanceof IdempotencyKeyReuse) || !state.jobId) throw err;
  job = await client.jobs.get(state.jobId);
}

client.jobs.get(job_id) 是 SDK 的重建路径，它需要一个 id，这正是为什么在提交返回的那一刻记录 id 才让这一切具备可恢复性。关于 POST /api/v2/jobs 上完整的键契约，请参阅 Comfy API v2。

[!NOTE] Comfy Router 的 Idempotency-Key 行为不同。在 Router 上，该键是一个重放句柄，而不是一次性令牌：用同一个键重新发送一个已排队的提交，会返回原始请求，而不是再排队一个。请参阅队列投递和 Router 重试结果。这两个接口只是共用一个请求头名称，契约并不相同。

查看任务运行

job.events() 会提供任务状态的实时流：节点和步骤进度、预览帧，以及每个输出一经提交的即时推送。如果连接中断，它会自动重新连接。

from comfy_sdk import Progress, Preview, OutputReady, StatusChange

job = client.submit(wf)

for event in job.events():
    match event:
        case Progress() as p:
            print(f"{p.value:.0%} {p.message}")
        case Preview() as pv:
            image = pv.to_pil()
        case OutputReady() as o:
            o.output.to_file(f"partial/{o.output.name}")
        case StatusChange(status="succeeded"):
            break

result = job.result()

const job = await client.submit(wf);

// 标签是必需的：switch 内的裸 `break` 跳出的是 switch，
// 而不是循环。
eventLoop: for await (const event of job.events()) {
  switch (event.kind) {
    case "progress":
      console.log(event.value);
      break;
    case "outputReady":
      await event.output.toFile(`${event.output.name}`);
      break;
    case "statusChange":
      if (event.status === "succeeded") break eventLoop;
  }
}

Preview.to_pil() 需要可选的 Pillow extra：pip install "comfy-sdk[pil]"。

result() 返回已完成的任务，如果执行失败，则抛出带有节点级详细信息的 JobFailed 异常。有关完整的事件目录，请参阅适用于您的语言的 SDK README。

events() 不是 subscribe()

在这两个接口中存在三个名称相似的东西，而本页只涉及其中一个：

Comfy Cloud 客户端上没有 subscribe()，任何地方也都没有 job.subscribe()。在 Cloud 上，subscribe 的等价物是 run()：提交并等待终止状态。

该流是实时信息流，而不是可重放的日志。它的存在是为了让你能够呈现进度，而不是让你依赖它来获取结果。轮询任务才是权威来源，run()、wait() 和 result() 会自动回退到轮询。原因请参阅 设计说明。

将输出回溯到其工作流

输出带有生成它们的任务的 ID，因此您可以以文件为起点反向追溯，而无需额外维护一张对应表。

output = job.outputs[0]
output.job_id          # the job that produced this file

const output = job.outputs[0];
output.jobId; // the job that produced this file

单独获取的资源也带有相同的 ID，因此您之后找到的文件依然能回溯到生成它的任务。对于您上传的资源，该 ID 为 None（TypeScript 中为 undefined），因为上传的资源没有对应的生成任务。

您可以向任务查询其背后的工作流。即使该任务不是在当前进程中提交的，只要通过 ID 重新加载，同样可以做到：

wf = job.get_workflow()

if wf.format == "save":
    ...  # the workflow as authored, canvas layout and Note nodes intact
else:
    ...  # the executed API-format graph

const wf = await job.getWorkflow();

if (wf.format === "save") {
  // the workflow as authored, canvas layout and Note nodes intact
} else {
  // the executed API-format graph
}

始终根据 format 进行分支判断。 返回哪种形状取决于任务的提交方式，而不是您在每次请求中能控制的任何因素：

您通过 SDK 提交的任务始终返回 api，因为 v2 提交目前还没有版本固定字段。这种情况将来会改变；判别字段的存在就是为了让您的代码无需随之改变。

为任务添加标签并在之后查找

标签是提交任务时附加的一组键和值，例如 client: acme 或 project: spring。用标签记录任务是为哪个客户或项目运行的。这样无需自己维护任务 ID 对照表，也能列出某个客户的任务。标签和任务列表需要当前的 SDK 版本，即安装中版本说明所列的版本。

标签适用于发送到 Comfy API 部署的任务，因此请按照选择基础 URL 中的说明，将 COMFY_BASE_URL 设为该部署的地址。Comfy Cloud 目前会拒绝标签，并对任务列表返回 501 和代码 not_implemented。自托管的 comfy-api-proxy 也不保存标签：它自己的 metadata 字段是单个文本字符串，因此会拒绝标签映射。它的任务列表会忽略标签筛选，因此在其上带标签的 SDK 列表调用不会返回任何任务，而通过 HTTP 发送相同的筛选条件会返回所有任务。

限制：

• 每个任务最多 16 个标签。
• 键为 1 到 40 个字符，只能使用字母、数字、_、- 和 .。
• 值是最多 256 字节（UTF-8）的字符串。
• 标签在提交任务时设置，之后不能更改。

违反限制的提交会以 422 和代码 metadata_invalid 失败，且不会创建任务。错误消息会指出违反规则的键；标签超过 16 个时，会给出标签数量。SDK 本身不检查这些限制，而是抛出一个带有服务器代码和消息的错误。

[!WARNING] 工作区中能看到某个任务的任何人，也能看到它的标签。请勿在标签中放入密钥或个人数据。

提交时传入 metadata。读回任务时，任务会带有这些标签：

job = client.submit(wf, metadata={"client": "acme", "project": "spring"})
job.metadata  # {"client": "acme", "project": "spring"}

const job = await client.submit(wf, {
  metadata: { client: "acme", project: "spring" },
});
job.metadata; // { client: "acme", project: "spring" }

未带标签提交的任务，其 metadata 为空。

要查找某个客户的任务，请按标签筛选并列出任务。列表只包含发送到基础 URL 所指部署的任务。结果按从新到旧排列，迭代时 SDK 会自动获取下一页：

for job in client.list_jobs(metadata={"client": "acme"}):
    print(job.id, job.status)

for await (const job of client.listJobs({ metadata: { client: "acme" } })) {
  console.log(job.id, job.status);
}

一次最多可按 3 个标签筛选，任务必须与所有标签完全匹配。列表项不包含输出。要获取任务的输出，请用 client.jobs.get(job.id) 按 ID 读取该任务。

通过 HTTP

在 POST /api/v2/jobs 的请求体中发送 metadata：

curl -X POST "https://<deployment>.run.comfy.app/api/v2/jobs" \
  -H "Authorization: Bearer $COMFY_API_KEY" \
  -H "Content-Type: application/json" \
  -d '{
    "workflow": { ... },
    "metadata": { "client": "acme", "project": "spring" }
  }'

筛选 GET /api/v2/jobs 时，每个标签对应一个 metadata[<key>]=<value> 查询参数。-g 参数可防止 curl 把方括号解析为匹配模式：

curl -g "https://<deployment>.run.comfy.app/api/v2/jobs?metadata[client]=acme" \
  -H "Authorization: Bearer $COMFY_API_KEY"

响应包含一页任务；如果还有更多页，还会包含 next_cursor。将它作为 cursor，连同相同的 metadata[...] 筛选条件一起传回，即可获取下一页。不带筛选条件发送的游标会返回未经筛选的任务。用 limit 设置每页大小（最多 500）。

{
  "jobs": [
    {
      "id": "6f1c2b9e-...",
      "status": "succeeded",
      "metadata": { "client": "acme", "project": "spring" }
    }
  ],
  "next_cursor": "eyJ..."
}

标签筛选超过 3 个，或键不符合键规则时，会以 400 和代码 invalid_metadata_filter 失败。任务和列表的全部字段请参阅 Comfy API v2 参考。

SDK 目前涵盖的内容

第一个版本只做好一件事：运行工作流并取回结果。

• 资产。 从文件、字节、流或 URL 创建输入句柄。句柄是惰性的，并按内容寻址，因此使用相同输入重新运行时不会再次上传。
• 提交。 提交 API 格式的节点图。提交是幂等的，队列已满时会在有限的预算内自动重试。
• 执行。 使用 wait() 轮询，或通过 events() 跟踪实时进度。
• 输出。 写入磁盘、缓冲到内存、获取字节范围，或获取短期有效的下载 URL。getDownloadUrl() 返回一个签名 URL，任何人都可以在没有 API 密钥的情况下读取，有效期约为 6 小时。在存储它之前请先阅读输出 URL 及其有效时长：若要在之后继续展示某个输出，请重新托管字节内容，或按需重新生成该 URL。
• 可追溯性。 每个输出都带有生成它的任务的 ID，并且任务可以返回其背后的工作流。
• 删除资产。 通过句柄或 ID 移除你上传的资产。
• 错误。 提供类型化异常，如 JobFailed、Unauthorized、InsufficientCredits 和 QueueFull，而不是原始状态码。
• 取消。 任务可以在运行时取消。TypeScript 还在任何调用上接受 AbortSignal。

Python 同时提供同步的 Comfy 客户端和接口相同的 AsyncComfy 客户端。TypeScript 仅支持异步。

此版本不包含：管理已保存的工作流、模型库、节点内省，以及命名工作流参数。设计说明 解释了为何接口从如此小的范围起步。

参考

SDK 的 README 是每种语言的完整参考，涵盖认证、资产、错误处理以及低层逃生通道。

[card] Python SDK comfy-sdk 位于 PyPI。提供同步和异步客户端。

[card] TypeScript SDK @comfyorg/sdk 位于 npm。带类型、异步，并提供低层客户端。

[card] Comfy API v2 参考 两个 SDK 底层的 HTTP API。任何语言都可以直接使用。

[card] 设计说明 这个 API 存在的原因、它与现有 ComfyUI API 的关系，以及未来的发展方向。

[card] Comfy Router 同一个包中的另一个接口：针对 Flux、Veo 和 Gemini 等合作伙伴模型的 comfy.models.run。

反馈

我们欢迎你对 SDK 和 API 提供反馈。请告诉我们哪些地方使用不便、哪些信息难以找到，以及你需要采用哪些变通方法。

请告诉我们哪些地方用起来别扭、你期望找到却没有找到的功能，以及你不得不绕行处理的地方。我们的 Discord 中的 #developer-platform 通道就是反馈这些内容的地方。

如果你希望有其他语言的第一方 SDK，也可以在那里提出。两个 SDK 都建立在同一套有文档记录的 HTTP 契约之上，所以如今任何语言都可以与 API 通信，但我们更希望了解需求在哪里。