ComfyUI - UI + Kit = ComfyKit
Python SDK for ComfyUI - Support Local or Cloud - Generate images, videos, audio in 3 lines
ComfyKit is a pure Python SDK that provides a clean API for executing ComfyUI workflows and returns structured Python objects.
from comfykit import ComfyKit
# Connect to local ComfyUI server
kit = ComfyKit(comfyui_url="http://127.0.0.1:8188")
result = await kit.execute("workflow.json", {"prompt": "a cute cat"})
print(result.images) # ['http://127.0.0.1:8188/view?filename=cat_001.png']
# 🌐 Or use RunningHub cloud (no local GPU needed)
# kit = ComfyKit(runninghub_api_key="rh-xxx")# ExecuteResult object, not strings!
result.status # "completed"
result.images # All generated image URLs
result.images_by_var # Images grouped by variable name
result.videos # Video URLs (if any)
result.audios # Audio URLs (if any)
result.duration # Execution time- ⚡ Zero Configuration: Works out of the box, connects to local ComfyUI by default (
http://127.0.0.1:8188) - ☁️ Cloud Execution: Seamless RunningHub cloud support - No GPU or local ComfyUI needed
- 🎨 Simple API: 3 lines of code to execute workflows, no need to understand internals
- 📊 Structured Output: Returns
ExecuteResultobjects, not strings - 🔄 Smart Detection: Auto-detects local files, URLs, and RunningHub workflow IDs
- 🔌 Lightweight: Less than 10 core dependencies
- 🎭 Multimodal Support: Images, videos, audio - all in one place
pip install comfykituv add comfykitIf you don't have a local GPU or ComfyUI environment, use RunningHub cloud:
import asyncio
from comfykit import ComfyKit
async def main():
# Initialize with RunningHub (only API key needed)
kit = ComfyKit(
runninghub_api_key="your-runninghub-key"
)
# Execute with workflow ID
result = await kit.execute("12345", {
"prompt": "a beautiful sunset over the ocean"
})
print(f"🖼️ Generated images: {result.images}")
asyncio.run(main())💡 Tip: Get your free API key at RunningHub
If you have ComfyUI running locally:
# Start ComfyUI (default port 8188)
python main.pyimport asyncio
from comfykit import ComfyKit
async def main():
# Connect to local ComfyUI (default: http://127.0.0.1:8188)
kit = ComfyKit(comfyui_url="http://127.0.0.1:8188")
# Execute workflow
result = await kit.execute(
"workflow.json",
params={"prompt": "a cute cat playing with yarn"}
)
# Check results
if result.status == "completed":
print(f"✅ Success! Duration: {result.duration:.2f}s")
print(f"🖼️ Images: {result.images}")
else:
print(f"❌ Failed: {result.msg}")
asyncio.run(main())💡 Tip:
comfyui_urldefaults tohttp://127.0.0.1:8188and can be omitted
from comfykit import ComfyKit
# Connect to local ComfyUI
kit = ComfyKit(comfyui_url="http://127.0.0.1:8188") # Default, can be omitted
# Execute local workflow file
result = await kit.execute("workflow.json", {
"prompt": "a cat",
"seed": 42,
"steps": 20
})# Connect to remote ComfyUI server
kit = ComfyKit(
comfyui_url="http://my-server:8188",
api_key="your-api-key" # If authentication is required
)# Use RunningHub cloud (no local ComfyUI needed)
kit = ComfyKit(
runninghub_api_key="your-runninghub-key"
)
# Execute with workflow ID
result = await kit.execute("12345", {
"prompt": "a beautiful landscape"
})# Automatically download and execute
result = await kit.execute(
"https://example.com/workflow.json",
{"prompt": "a cat"}
)workflow_dict = {
"nodes": [...],
"edges": [...]
}
result = await kit.execute_json(workflow_dict, {
"prompt": "a cat"
})result = await kit.execute("workflow.json", {"prompt": "a cat"})
# Basic info
print(f"Status: {result.status}") # completed / failed
print(f"Duration: {result.duration}s") # 3.45
print(f"Prompt ID: {result.prompt_id}") # uuid
# Generated media files
print(f"Images: {result.images}") # ['http://...']
print(f"Videos: {result.videos}") # ['http://...']
print(f"Audios: {result.audios}") # ['http://...']
# Grouped by variable name (if workflow defines output variables)
print(f"Cover: {result.images_by_var['cover']}")
print(f"Thumbnail: {result.images_by_var['thumbnail']}")ComfyKit provides a concise DSL (Domain Specific Language) for marking workflow nodes, allowing you to:
- Define dynamic parameters
- Mark output variables
- Specify required/optional parameters
- Automatically handle media file uploads
These DSL markers are written in the title field of ComfyUI workflow nodes to convert fixed workflows into parameterizable templates.
Usage Steps:
- In ComfyUI editor, double-click a node and modify its title to add DSL markers (e.g.,
$prompt.text!) - Save as API format JSON (select "Save (API Format)" from menu, not regular "Save")
- Execute with parameters via
kit.execute("workflow.json", {"prompt": "value"})
⚠️ Important: ComfyKit requires API format workflow JSON, not UI format.
| Syntax | Description | Example | Effect |
|---|---|---|---|
$param |
Basic parameter (shorthand) | $prompt |
Parameter prompt, maps to field prompt |
$param.field |
Specify field mapping | $prompt.text |
Parameter prompt, maps to field text |
$param! |
Required parameter | $prompt! |
Parameter prompt is required, no default |
$~param |
Media parameter (upload) | $~image |
Parameter image requires file upload |
$~param! |
Required media parameter | $~image! |
Parameter image is required and needs upload |
$param.~field! |
Combined markers | $img.~image! |
Parameter img maps to image, required and upload |
$output.name |
Output variable marker | $output.cover |
Mark output variable name as cover |
Text, $p1, $p2 |
Multiple parameters | Size, $width!, $height! |
Define multiple parameters in one node |
In a ComfyUI workflow CLIPTextEncode node:
{
"6": {
"class_type": "CLIPTextEncode",
"_meta": {
"title": "$prompt.text!"
},
"inputs": {
"text": "a beautiful landscape",
"clip": ["4", 1]
}
}
}Marker explanation:
$prompt- Parameter name isprompt.text- Maps to node'stextfield!- Required parameter, must be provided
Usage:
result = await kit.execute("workflow.json", {
"prompt": "a cute cat" # Replaces inputs.text value
})In a LoadImage node:
{
"10": {
"class_type": "LoadImage",
"_meta": {
"title": "$~input_image!"
},
"inputs": {
"image": "default.png"
}
}
}Marker explanation:
$~input_image!- Parameterinput_image, needs upload (~), required (!)- ComfyKit handles file upload automatically
Usage:
result = await kit.execute("workflow.json", {
"input_image": "/path/to/cat.jpg" # Automatically uploads to ComfyUI
}){
"5": {
"class_type": "EmptyLatentImage",
"_meta": {
"title": "Size, $width!, $height!"
},
"inputs": {
"width": 512,
"height": 512,
"batch_size": 1
}
}
}Marker explanation:
Size- Display text, not a parameter$width!- Required parameterwidth(shorthand, maps to same field)$height!- Required parameterheight
Usage:
result = await kit.execute("workflow.json", {
"width": 1024,
"height": 768
}){
"3": {
"class_type": "KSampler",
"_meta": {
"title": "Sampler, $seed, $steps"
},
"inputs": {
"seed": 0, # Default value 0
"steps": 20, # Default value 20
"cfg": 8.0,
"model": ["4", 0]
}
}
}Marker explanation:
$seedand$stepshave no!, they are optional- If not provided, uses default values from workflow
Usage:
# Use defaults
result = await kit.execute("workflow.json", {})
# Override some parameters
result = await kit.execute("workflow.json", {
"seed": 42 # Only override seed, steps uses default 20
}){
"9": {
"class_type": "SaveImage",
"_meta": {
"title": "$output.cover"
},
"inputs": {
"filename_prefix": "book_cover",
"images": ["8", 0]
}
}
}Marker explanation:
$output.cover- Mark this node's output ascovervariable
Usage:
result = await kit.execute("workflow.json", params)
# Access output by variable name
cover_images = result.images_by_var["cover"]
print(f"Cover image: {cover_images[0]}"){
"9": {
"class_type": "SaveImage",
"_meta": {
"title": "$output.cover"
}
},
"15": {
"class_type": "SaveImage",
"_meta": {
"title": "$output.thumbnail"
}
}
}Usage:
result = await kit.execute("workflow.json", params)
# Get different outputs separately
cover = result.images_by_var["cover"][0]
thumbnail = result.images_by_var["thumbnail"][0]If you don't use $output.xxx markers, ComfyKit auto-detects output nodes:
{
"9": {
"class_type": "SaveImage",
"_meta": {
"title": "Final Output"
}
}
}Usage:
result = await kit.execute("workflow.json", params)
# All images are in the images list
all_images = result.images
# Access by node ID
images_from_node_9 = result.images_by_var["9"]- Parameter Naming: Use descriptive names like
$positive_promptinstead of$p - Required Markers: Use
!for parameters with no reasonable default - Upload Markers: Use
~for image, video, audio parameters - Output Variables: Use
$output.xxxfor important outputs to make them easy to reference - Display Text: Add descriptive text in multi-param markers, e.g.
"Size, $width!, $height!"
A complete Text-to-Image workflow with DSL markers:
{
"4": {
"class_type": "CheckpointLoaderSimple",
"_meta": {
"title": "$model.ckpt_name"
},
"inputs": {
"ckpt_name": "sd_xl_base_1.0.safetensors"
}
},
"5": {
"class_type": "EmptyLatentImage",
"_meta": {
"title": "Canvas, $width!, $height!"
},
"inputs": {
"width": 1024,
"height": 1024,
"batch_size": 1
}
},
"6": {
"class_type": "CLIPTextEncode",
"_meta": {
"title": "$prompt.text!"
},
"inputs": {
"text": "a beautiful landscape",
"clip": ["4", 1]
}
},
"9": {
"class_type": "SaveImage",
"_meta": {
"title": "$output.result"
},
"inputs": {
"filename_prefix": "output",
"images": ["8", 0]
}
}
}Execution:
result = await kit.execute("t2i_workflow.json", {
"prompt": "a cute cat playing with yarn",
"width": 1024,
"height": 768,
"model": "dreamshaper_8.safetensors" # Optional, has default
})
# Get result
output_image = result.images_by_var["result"][0]ComfyKit uses the following priority for configuration:
- Constructor parameters (highest priority)
- Environment variables
- Default values
kit = ComfyKit(
# ComfyUI server URL
comfyui_url="http://127.0.0.1:8188", # Default
# Execution mode: http (recommended) or websocket
executor_type="http", # Default
# API Key (if ComfyUI requires authentication)
api_key="your-api-key",
# Cookies (if needed)
cookies="session=abc123"
)kit = ComfyKit(
# RunningHub API URL
runninghub_url="https://www.runninghub.ai", # Default
# RunningHub API Key (required)
runninghub_api_key="rh-key-xxx",
# Timeout (seconds)
runninghub_timeout=300, # Default: 5 minutes
# Retry count
runninghub_retry_count=3 # Default: 3 retries
)# ComfyUI configuration
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 configuration
export RUNNINGHUB_BASE_URL="https://www.runninghub.ai"
export RUNNINGHUB_API_KEY="rh-key-xxx"
export RUNNINGHUB_TIMEOUT="300"
export RUNNINGHUB_RETRY_COUNT="3"| Aspect | ComfyUI Native API | ComfyKit |
|---|---|---|
| Complexity | Manual WebSocket/HTTP handling | 3 lines of code |
| Return Value | Raw JSON, need to parse yourself | Structured ExecuteResult object |
| Media Handling | Need to construct URLs manually | Automatically generates complete media URLs |
| Error Handling | Need to implement yourself | Built-in comprehensive error handling |
| Best For | Familiar with ComfyUI internals | Just want quick integration |
class ComfyKit:
def __init__(
self,
# Local ComfyUI configuration
comfyui_url: Optional[str] = None,
executor_type: Literal["http", "websocket"] = "http",
api_key: Optional[str] = None,
cookies: Optional[str] = None,
# RunningHub cloud configuration
runninghub_url: Optional[str] = None,
runninghub_api_key: Optional[str] = None,
runninghub_timeout: int = 300,
runninghub_retry_count: int = 3,
):
"""Initialize ComfyKit
All parameters are optional and can be configured via environment variables
"""
async def execute(
self,
workflow: Union[str, Path],
params: Optional[Dict[str, Any]] = None,
) -> ExecuteResult:
"""Execute workflow
Args:
workflow: Workflow source, can be:
- Local file path: "workflow.json"
- RunningHub ID: "12345" (numeric)
- Remote URL: "https://example.com/workflow.json"
params: Workflow parameters, e.g. {"prompt": "a cat", "seed": 42}
Returns:
ExecuteResult: Structured execution result
"""
async def execute_json(
self,
workflow_json: Dict[str, Any],
params: Optional[Dict[str, Any]] = None,
) -> ExecuteResult:
"""Execute workflow from JSON dict
Args:
workflow_json: Workflow JSON dict
params: Workflow parameters
Returns:
ExecuteResult: Structured execution result
"""class ExecuteResult:
"""Workflow execution result"""
status: str # Execution status: "completed" / "failed"
prompt_id: Optional[str] # Prompt ID
duration: Optional[float] # Execution duration (seconds)
# Media outputs
images: List[str] # All image URLs
videos: List[str] # All video URLs
audios: List[str] # All audio URLs
texts: List[str] # All text outputs
# Grouped by variable name
images_by_var: Dict[str, List[str]] # Images grouped by variable name
videos_by_var: Dict[str, List[str]] # Videos grouped by variable name
audios_by_var: Dict[str, List[str]] # Audios grouped by variable name
texts_by_var: Dict[str, List[str]] # Texts grouped by variable name
# Raw outputs
outputs: Optional[Dict[str, Any]] # Raw output data
msg: Optional[str] # Error message (if failed)The project includes complete example code in the examples/ directory:
01_quick_start.py- Quick start guide02_configuration.py- Configuration options03_local_workflows.py- Local workflow execution04_runninghub_cloud.py- RunningHub cloud execution05_advanced_features.py- Advanced features
Run all examples:
cd examples
python run_all.pyuv sync --extra devpytestruff check --fix
ruff formatContributions are welcome! Please check Issues for areas that need help.
- Fork this repository
- Create a feature branch (
git checkout -b feature/amazing-feature) - Commit your changes (
git commit -m 'Add amazing feature') - Push to the branch (
git push origin feature/amazing-feature) - Open a Pull Request
This project is licensed under the MIT License - see the LICENSE file for details.
- ComfyUI - Powerful AI image generation framework
- RunningHub - ComfyUI cloud platform
- Author: Fan Wu
- Email: 1129090915@qq.com
- GitHub: @puke3615