comfykit 0.1.0

ComfyUI Python SDK for developers - Build AI generation into your app

ComfyKit

ComfyUI - UI + Kit = ComfyKit

面向开发者的 ComfyUI Python SDK，让 AI 图像生成集成变得简单

---

🤔 为什么需要 ComfyKit？

如果你在用 ComfyUI 做 AI 图像生成，可能遇到过这些问题：

• 🔧 ComfyUI 原生 API 太底层：需要手动处理 WebSocket、轮询任务状态、解析复杂的返回格式
• 📝 返回的是字符串，不是结构化数据：难以集成到项目中
• ⚙️ 配置复杂：需要了解 ComfyUI 内部机制才能正确调用
• 💻 本地部署门槛高：需要高性能 GPU 和复杂的环境配置
• 🔄 代码重复：每个项目都要重新实现一遍 workflow 执行逻辑

ComfyKit 就是来解决这些问题的。

---

✨ ComfyKit 是什么？

ComfyKit 是一个纯粹的 Python SDK，提供简洁的 API 来执行 ComfyUI workflows，返回结构化的 Python 对象。

3 行代码执行一个 workflow

from comfykit import ComfyKit

kit = ComfyKit()
result = await kit.execute("workflow.json", {"prompt": "a cute cat"})

print(result.images)  # ['http://127.0.0.1:8188/view?filename=cat_001.png']

获得结构化的返回数据

# ExecuteResult 对象，不是字符串！
result.status          # "completed"
result.images          # 所有生成的图片 URL
result.images_by_var   # 按变量名分组的图片
result.videos          # 视频 URL（如果有）
result.audios          # 音频 URL（如果有）
result.duration        # 执行耗时

---

🎯 核心特性

• ⚡ 开箱即用：零配置，默认连接本地 ComfyUI（http://127.0.0.1:8188）
• ☁️ 云端执行：无缝支持 RunningHub 云平台，无需本地 GPU 和 ComfyUI 环境
• 🎨 简洁 API：3 行代码执行 workflow，无需了解底层细节
• 📊 结构化返回：返回 ExecuteResult 对象，不是字符串
• 🔄 智能识别：自动识别本地文件、URL、RunningHub workflow ID
• 🔌 最小依赖：核心依赖少于 10 个，轻量级
• 🎭 多模态支持：图片、视频、音频一站式处理

---

📦 安装

使用 pip

pip install comfykit

使用 uv（推荐）

uv add comfykit

---

🚀 快速开始

ComfyKit 支持两种执行方式：本地 ComfyUI 和 RunningHub 云端。

方式 1：本地 ComfyUI（需要本地环境）

1. 启动 ComfyUI

# 启动 ComfyUI（默认端口 8188）
python main.py

2. 准备一个 workflow 文件

{
  "3": {
    "class_type": "KSampler",
    "inputs": {
      "seed": 0,
      "steps": 20,
      "cfg": 8.0,
      "sampler_name": "euler",
      "scheduler": "normal",
      "denoise": 1.0,
      "model": ["4", 0],
      "positive": ["6", 0],
      "negative": ["7", 0],
      "latent_image": ["5", 0]
    }
  },
  "6": {
    "class_type": "CLIPTextEncode",
    "inputs": {
      "text": "a beautiful landscape",
      "clip": ["4", 1]
    }
  }
  // ... more nodes
}

3. 执行 workflow

import asyncio
from comfykit import ComfyKit

async def main():
    # 初始化（使用默认配置）
    kit = ComfyKit()
    
    # 执行 workflow
    result = await kit.execute(
        "workflow.json",
        params={"prompt": "a cute cat playing with yarn"}
    )
    
    # 查看结果
    if result.status == "completed":
        print(f"✅ 生成成功！耗时 {result.duration:.2f}s")
        print(f"🖼️  生成的图片：{result.images}")
    else:
        print(f"❌ 生成失败：{result.msg}")

asyncio.run(main())

方式 2：RunningHub 云端（无需本地环境）⭐

如果你没有本地 GPU 或 ComfyUI 环境，可以使用 RunningHub 云平台：

import asyncio
from comfykit import ComfyKit

async def main():
    # 初始化云端执行（只需 API Key）
    kit = ComfyKit(
        runninghub_api_key="your-runninghub-key"
    )
    
    # 使用 RunningHub workflow ID 执行
    result = await kit.execute("12345", {
        "prompt": "a beautiful landscape"
    })
    
    print(f"🖼️  生成的图片：{result.images}")

asyncio.run(main())

💡 提示：访问 RunningHub 获取免费 API Key

---

📚 使用示例

本地 ComfyUI 执行

from comfykit import ComfyKit

# 默认连接本地 ComfyUI
kit = ComfyKit()

# 执行本地 workflow 文件
result = await kit.execute("workflow.json", {
    "prompt": "a cat",
    "seed": 42,
    "steps": 20
})

自定义 ComfyUI 地址

# 连接到远程 ComfyUI 服务器
kit = ComfyKit(
    comfyui_url="http://my-server:8188",
    api_key="your-api-key"  # 如果需要认证
)

RunningHub 云端执行

# 使用 RunningHub 云平台（无需本地 ComfyUI）
kit = ComfyKit(
    runninghub_api_key="your-runninghub-key"
)

# 使用 workflow ID 执行
result = await kit.execute("12345", {
    "prompt": "a beautiful landscape"
})

执行远程 workflow URL

# 自动下载并执行
result = await kit.execute(
    "https://example.com/workflow.json",
    {"prompt": "a cat"}
)

执行 workflow JSON 字典

workflow_dict = {
    "nodes": [...],
    "edges": [...]
}

result = await kit.execute_json(workflow_dict, {
    "prompt": "a cat"
})

处理返回结果

result = await kit.execute("workflow.json", {"prompt": "a cat"})

# 基本信息
print(f"状态: {result.status}")           # completed / failed
print(f"耗时: {result.duration}秒")       # 3.45
print(f"Prompt ID: {result.prompt_id}")   # uuid

# 生成的媒体文件
print(f"图片: {result.images}")           # ['http://...']
print(f"视频: {result.videos}")           # ['http://...']
print(f"音频: {result.audios}")           # ['http://...']

# 按变量名分组（如果 workflow 定义了输出变量）
print(f"封面图: {result.images_by_var['cover']}")
print(f"缩略图: {result.images_by_var['thumbnail']}")

---

⚙️ 配置说明

配置优先级

ComfyKit 使用以下优先级读取配置：

1. 代码传参（最高优先级）
2. 环境变量
3. 默认值

ComfyUI 本地执行配置

kit = ComfyKit(
    # ComfyUI 服务器地址
    comfyui_url="http://127.0.0.1:8188",  # 默认值
    
    # 执行模式：http（推荐）或 websocket
    executor_type="http",  # 默认值
    
    # API Key（如果 ComfyUI 开启了认证）
    api_key="your-api-key",
    
    # Cookies（如果需要）
    cookies="session=abc123"
)

RunningHub 云端执行配置

kit = ComfyKit(
    # RunningHub API 地址
    runninghub_url="https://www.runninghub.ai",  # 默认值
    
    # RunningHub API Key（必需）
    runninghub_api_key="rh-key-xxx",
    
    # 超时时间（秒）
    runninghub_timeout=300,  # 默认 5 分钟
    
    # 重试次数
    runninghub_retry_count=3  # 默认 3 次
)

环境变量配置

# ComfyUI 配置
export COMFYUI_BASE_URL="http://127.0.0.1:8188"
export COMFYUI_EXECUTOR_TYPE="http"
export COMFYUI_API_KEY="your-api-key"
export COMFYUI_COOKIES="session=abc123"

# RunningHub 配置
export RUNNINGHUB_BASE_URL="https://www.runninghub.ai"
export RUNNINGHUB_API_KEY="rh-key-xxx"
export RUNNINGHUB_TIMEOUT="300"
export RUNNINGHUB_RETRY_COUNT="3"

---

🔍 ComfyKit vs ComfyUI 原生 API

维度	ComfyUI 原生 API	ComfyKit
复杂度	需要手动处理 WebSocket/HTTP	3 行代码执行
返回值	原始 JSON，需要自己解析	结构化 ExecuteResult 对象
媒体处理	需要手动拼接 URL	自动生成完整的媒体 URL
云端支持	不支持	内置 RunningHub 云端执行
错误处理	需要自己实现	内置完善的错误处理
适合人群	熟悉 ComfyUI 内部机制	只想快速集成

---

📖 API 参考

ComfyKit 类

class ComfyKit:
    def __init__(
        self,
        # ComfyUI 本地执行配置
        comfyui_url: Optional[str] = None,
        executor_type: Literal["http", "websocket"] = "http",
        api_key: Optional[str] = None,
        cookies: Optional[str] = None,
        
        # RunningHub 云端执行配置
        runninghub_url: Optional[str] = None,
        runninghub_api_key: Optional[str] = None,
        runninghub_timeout: int = 300,
        runninghub_retry_count: int = 3,
    ):
        """初始化 ComfyKit
        
        所有参数都是可选的，可以通过环境变量配置
        """
        
    async def execute(
        self,
        workflow: Union[str, Path],
        params: Optional[Dict[str, Any]] = None,
    ) -> ExecuteResult:
        """执行 workflow
        
        Args:
            workflow: workflow 来源，可以是：
                     - 本地文件路径: "workflow.json"
                     - RunningHub ID: "12345"（纯数字）
                     - 远程 URL: "https://example.com/workflow.json"
            params: workflow 参数，例如 {"prompt": "a cat", "seed": 42}
        
        Returns:
            ExecuteResult: 结构化的执行结果
        """
        
    async def execute_json(
        self,
        workflow_json: Dict[str, Any],
        params: Optional[Dict[str, Any]] = None,
    ) -> ExecuteResult:
        """从 JSON 字典执行 workflow
        
        Args:
            workflow_json: workflow JSON 字典
            params: workflow 参数
        
        Returns:
            ExecuteResult: 结构化的执行结果
        """

ExecuteResult 类

class ExecuteResult:
    """Workflow 执行结果"""
    
    status: str                           # 执行状态: "completed" / "failed"
    prompt_id: Optional[str]              # Prompt ID
    duration: Optional[float]             # 执行耗时（秒）
    
    # 媒体输出
    images: List[str]                     # 所有图片 URL
    videos: List[str]                     # 所有视频 URL
    audios: List[str]                     # 所有音频 URL
    texts: List[str]                      # 所有文本输出
    
    # 按变量名分组的输出
    images_by_var: Dict[str, List[str]]   # 图片按变量名分组
    videos_by_var: Dict[str, List[str]]   # 视频按变量名分组
    audios_by_var: Dict[str, List[str]]   # 音频按变量名分组
    texts_by_var: Dict[str, List[str]]    # 文本按变量名分组
    
    # 原始输出
    outputs: Optional[Dict[str, Any]]     # 原始输出数据
    msg: Optional[str]                    # 错误消息（如果失败）

---

📂 更多示例

项目包含完整的示例代码，位于 examples/ 目录：

• 01_quick_start.py - 快速入门
• 02_configuration.py - 配置选项
• 03_local_workflows.py - 本地 workflow 执行
• 04_runninghub_cloud.py - RunningHub 云端执行
• 05_advanced_features.py - 高级特性

运行所有示例：

cd examples
python run_all.py

---

🛠️ 开发

安装开发依赖

uv sync --extra dev

运行测试

pytest

代码格式化

ruff check --fix
ruff format

---

🤝 贡献

欢迎贡献！请查看 Issues 了解当前需要帮助的地方。

贡献流程

1. Fork 本仓库
2. 创建特性分支 (git checkout -b feature/amazing-feature)
3. 提交更改 (git commit -m 'Add amazing feature')
4. 推送到分支 (git push origin feature/amazing-feature)
5. 开启 Pull Request

---

📄 许可证

本项目采用 MIT 许可证 - 查看 LICENSE 文件了解详情。

---

🙏 致谢

• ComfyUI - 强大的 AI 图像生成框架
• RunningHub - ComfyUI 云平台

---

📞 联系

• 作者：Fan Wu
• Email：[e-mail]
• GitHub：@puke3615

---

如果 ComfyKit 对你有帮助，请给个 ⭐ Star！

GitHub · PyPI · 问题反馈