ComfyUI API调用

导出工作流API的JSON文件

在进行API调用之前，请先在ComfyUI WebUI中搭建并调试您所需的工作流，并确认该工作流可以成功生成图像。 调试完毕后，在ComfyUI界面点击菜单项工作流 > 导出(API)，您的浏览器将会下载一个包含工作流定义的JSON文件。

该JSON文件描述了当前工作流是如何工作的。

{
  "3": {
    "inputs": {
      "seed": 740511732689542,
      "steps": 20,
      "cfg": 8,
      "sampler_name": "euler",
      "scheduler": "normal",
      "denoise": 1,
      "model": [
        "4",
        0
      ],
      "positive": [
        "6",
        0
      ],
      "negative": [
        "7",
        0
      ],
      "latent_image": [
        "5",
        0
      ]
    },
    "class_type": "KSampler",
    "_meta": {
      "title": "K采样器"
    }
  },
  "4": {
    "inputs": {
      "ckpt_name": "sd-v1-5-inpainting.ckpt"
    },
    "class_type": "CheckpointLoaderSimple",
    "_meta": {
      "title": "加载检查点"
    }
  },
  "5": {
    "inputs": {
      "width": 512,
      "height": 512,
      "batch_size": 1
    },
    "class_type": "EmptyLatentImage",
    "_meta": {
      "title": "空潜空间图像"
    }
  },
  "6": {
    "inputs": {
      "text": "beautiful scenery nature glass bottle landscape, , purple galaxy bottle,",
      "clip": [
        "4",
        1
      ]
    },
    "class_type": "CLIPTextEncode",
    "_meta": {
      "title": "CLIP文本编码（提示）"
    }
  },
  "7": {
    "inputs": {
      "text": "text, watermark",
      "clip": [
        "4",
        1
      ]
    },
    "class_type": "CLIPTextEncode",
    "_meta": {
      "title": "CLIP文本编码（提示）"
    }
  },
  "8": {
    "inputs": {
      "samples": [
        "3",
        0
      ],
      "vae": [
        "4",
        2
      ]
    },
    "class_type": "VAEDecode",
    "_meta": {
      "title": "VAE解码"
    }
  },
  "9": {
    "inputs": {
      "filename_prefix": "ComfyUI",
      "images": [
        "8",
        0
      ]
    },
    "class_type": "SaveImage",
    "_meta": {
      "title": "保存图像"
    }
  }
}

发布API

工作流JSON文件准备就绪后，接下来需要为您的ComfyUI函数配置API访问入口。切换到API调用页签，单击发布API。

修改GPU规格和弹性策略

在配置ComfyUI API接口时，您需要根据预期的调用负载和性能要求，合理设置函数的资源规格。主要考量以下参数：

• GPU规格：选择不同卡型和配置的GPU实例来处理API调用请求。不同的ComfyUI工作流，其计算复杂度和资源消耗差异巨大。我们建议您根据实际工作流进行测试，评估不同GPU规格在任务处理时间、成本上的表现，以选择最优配置。

• 弹性策略：

  • 按量模式：完全按量计费，没有请求的阶段不会产生计算费用，稀疏调用场景下会有冷启动，首个请求响应较慢。

  • 极速模式：将ComfyUI缓存至预置快照中，此时仅会产生极低的快照存储费用。通过这种方式，可以有效避免冷启动问题，显著提升低频调用的响应速度。当所需要的实例数超过预置快照的数量时，仍然可能会弹出按量实例，此时可能部分请求会经历冷启动的阶段。

完成API的发布配置后，平台会自动生成一个可用于外部调用的API地址。参照页面内的API调用文档，并结合前面导出的JSON文件，即可构造API请求，完成AI生图业务调用。

HTTP出图

WebSocket出图

该接口逻辑与HTTP模式一致，输入参数通过websocket.send()传入。

状态获取

获取出图任务状态，由于FaaS无状态的特性，为确保可以正确持久化状态信息，需要进行存储配置。

存储至文件系统：挂载NAS、OSS到函数实例，文件将会被存储至{挂载目录}/output/serverless_api。

同步调用出图

仅关心出图结果，不关心出图进度，可以通过同步HTTP调用。

import requests
import json

# 填入您的公网请求地址
endpoint = "https://xxx.xxx.fcapp.run"


def txt2img(prompt):
    resp = requests.post(f"{endpoint}/api/serverless/run?output_oss=true", json=prompt)

    return resp.json()


with open("./api.json", "r", encoding="utf-8") as f:
    result = txt2img(json.loads(f.read()))
    images = result["data"]["results"]
    for image in images:
        # 会输出 serverless-api-output/b74f33c0-69ba-4e4b-b584-6700890b7f51.png
        print(image["output"]["oss"]["object"])

流式出图

如果需要实时获取出图进度，可以考虑通过同步HTTP调用，同时读取流式输出。

import requests
import json

# 填入您的公网请求地址
endpoint = "https://xxx.xxx.fcapp.run"


def txt2img(prompt):

    with requests.post(
        f"{endpoint}/api/serverless/run?output_oss=true&stream=true",
        json=prompt,
        stream=True,
    ) as response:
        for line in response.iter_lines():
            # 只有 data: 开头的是数据行
            if line.startswith(b"data: "):
                data = json.loads(line[6:])
                # 随着出图进度的变化，会输出当前的节点状态
                # 部分可能出现的行如下
                # {'type': 'status', 'data': {'status': {'exec_info': {'queue_remaining': 1}}, 'sid': '200c62b4-111f-4994-a4a4-20a28c944408'}}
                # {'type': 'executing', 'data': {'node': '4'}}
                # {'type': 'status', 'data': {'status': {'exec_info': {'queue_remaining': 1}}}}
                # {'type': 'execution_start', 'data': {'prompt_id': '453484bb-1749-4fe3-9f2c-31664b8c3aec', 'timestamp': 1747821408300}}
                # {'type': 'execution_cached', 'data': {'nodes': [], 'prompt_id': '453484bb-1749-4fe3-9f2c-31664b8c3aec', 'timestamp': 1747821408304}}
                # {'type': 'executing', 'data': {'node': '4', 'display_node': '4', 'prompt_id': '453484bb-1749-4fe3-9f2c-31664b8c3aec'}}
                # {'type': 'progress', 'data': {'value': 1, 'max': 20, 'prompt_id': '453484bb-1749-4fe3-9f2c-31664b8c3aec', 'node': '3'}}
                # {'type': 'executing', 'data': {'node': None, 'prompt_id': '453484bb-1749-4fe3-9f2c-31664b8c3aec'}}
                # 'type': 'serverless_api', 'data': ... }
                print(data)

                # 最终结果的类型的是 serverless_api
                if data["type"] == "serverless_api":
                    return data


with open("./api.json", "r", encoding="utf-8") as f:
    result = txt2img(json.loads(f.read()))
    images = result["data"]["results"]
    for image in images:
        # 会输出 serverless-api-output/b74f33c0-69ba-4e4b-b584-6700890b7f51.png
        print(image["output"]["oss"]["object"])

异步调用出图

如果不希望使用时被阻塞住，可以通过函数计算自带的异步调用能力，并通过获取状态接口轮询获取结果。异步调用能力由函数计算提供，请参考 InvokeFunction - 调用函数。

from time import sleep
import requests
import json
import uuid

# 填入您的公网请求地址
endpoint = "https://xxx.xxx.fcapp.run"


def txt2img(prompt):
    task_id = str(uuid.uuid4())

    requests.post(
        f"{endpoint}/api/serverless/run?output_oss=true",
        json=prompt,
        headers={"X-Fc-Invocation-Type": "Async", "X-Fc-Trace-Id": task_id},
    )

    while True:
        resp = requests.get(
            f"{endpoint}/api/serverless/status?task_id={task_id}",
        )
        statusList = resp.json()
        # 会以数组的方式输出当前出图任务的所有中间状态
        print(statusList)

        if len(statusList) > 0:
            if statusList[-1]["type"] == "serverless_api":
                return statusList[-1]

        sleep(1)


with open("./api.json", "r", encoding="utf-8") as f:
    result = txt2img(json.loads(f.read()))
    images = result["data"]["results"]
    for image in images:
        # 会输出 serverless-api-output/b74f33c0-69ba-4e4b-b584-6700890b7f51.png
        print(image["output"]["oss"]["object"])

WebSocket出图

如果已经接入了WebSocket原生API，并通过WebSocket获取出图进度，可以使用WebSocket生图接口，在最小改动的情况下，使用ComfyUI Serverless API。

import json
import websocket  # pip install websocket-client

# 填入您的公网请求地址
# 注意要用 ws 或 wss 协议
endpoint = "wss://xxx.xxx.fcapp.run"


def txt2img(prompt):
    result = {}

    def on_open(ws):
        ws.send(json.dumps(prompt))

    def on_message(ws, msg):
        # 随着出图进度的变化，会输出当前的节点状态
        data = json.loads(msg)
        print(data)

        if data["type"] == "serverless_api":
            nonlocal result
            result = data
            ws.close()

    ws = websocket.WebSocketApp(
        f"{endpoint}/api/serverless/ws?output_oss=true",
        on_open=on_open,
        on_message=on_message,
    )

    ws.run_forever()

    return result


with open("./api.json", "r", encoding="utf-8") as f:
    result = txt2img(json.loads(f.read()))
    images = result["data"]["results"]
    for image in images:
        # 会输出 serverless-api-output/b74f33c0-69ba-4e4b-b584-6700890b7f51.png
        print(image["output"]["oss"]["object"])

工作流找不到了怎么办？

如果您已经出图成功，可以在左侧菜单找到出图历史记录，在队列中选择图片，单击右键，选择加载工作流进行恢复。