Build a ComfyUI-inspired workflow engine in 45 minutes
Welcome to another official UVF IT build guide. Today we are constructing a minimal, working clone of a node-based visual programming interface, inspired by the modular architecture of ComfyUI.
We will build a Python backend that defines a directed acyclic graph (DAG) of image processing nodes and a simple web UI to visualize and execute these workflows, using only free-tier open-source tools.
Create a virtual environment and install FastAPI for the API layer, Pillow for image processing, and Uvicorn for serving.
Initialize the project structure with a main server file and a directory for static assets.
mkdir comfy-clone && cd comfy-clone
python3 -m venv venv
source venv/bin/activate
pip install fastapi uvicorn pillow
mkdir staticCreate a Python module that defines a 'Node' class with inputs, outputs, and an execute method.
Implement a simple DAG executor that resolves dependencies and runs nodes in topological order.
We will use PIL to simulate image generation (e.g., creating a blank image or adding text) to avoid heavy ML dependencies.
engine.py
from PIL import Image, ImageDraw, ImageFilter
import io
import base64
from typing import Dict, Any, List
class Node:
def __init__(self, node_id: str, node_type: str, inputs: Dict[str, Any] = None):
self.node_id = node_id
self.node_type = node_type
self.inputs = inputs or {}
self.output = None
def execute(self, context: Dict[str, Any]) -> Any:
if self.node_type == "EmptyImage":
width = self.inputs.get("width", 256)
height = self.inputs.get("height", 256)
color = self.inputs.get("color", "white")
img = Image.new("RGB", (width, height), color)
return img
elif self.node_type == "Grayscale":
# Find the input image from context
input_img = None
for k, v in self.inputs.items():
if isinstance(v, Image.Image):
input_img = v
break
if input_img is None:
raise ValueError("Grayscale node requires an image input")
return input_img.convert("L")
elif self.node_type == "TextOverlay":
input_img = None
text = "Generated by UVF IT"
for k, v in self.inputs.items():
if isinstance(v, Image.Image):
input_img = v
if isinstance(v, str):
text = v
if input_img is None:
raise ValueError("TextOverlay node requires an image input")
draw = ImageDraw.Draw(input_img)
draw.text((10, 10), text, fill="black")
return input_img
else:
raise ValueError(f"Unknown node type: {self.node_type}")
class GraphEngine:
def __init__(self):
self.nodes: Dict[str, Node] = {}
self.edges: List[Dict[str, str]] = []
def add_node(self, node: Node):
self.nodes[node.node_id] = node
def add_edge(self, from_id: str, to_id: str, input_key: str):
self.edges.append({"from": from_id, "to": to_id, "input_key": input_key})
def execute(self) -> Dict[str, str]:
# Simple topological sort via Kahn's algorithm
in_degree = {node_id: 0 for node_id in self.nodes}
adj = {node_id: [] for node_id in self.nodes}
for edge in self.edges:
adj[edge["from"]].append((edge["to"], edge["input_key"]))
in_degree[edge["to"]] += 1
queue = [node_id for node_id, degree in in_degree.items() if degree == 0]
execution_order = []
while queue:
node_id = queue.pop(0)
execution_order.append(node_id)
for neighbor, input_key in adj[node_id]:
in_degree[neighbor] -= 1
if in_degree[neighbor] == 0:
queue.append(neighbor)
if len(execution_order) != len(self.nodes):
raise ValueError("Cycle detected in graph")
context = {}
results = {}
for node_id in execution_order:
node = self.nodes[node_id]
# Resolve inputs from context
resolved_inputs = {}
for k, v in node.inputs.items():
if isinstance(v, str) and v in context:
resolved_inputs[k] = context[v]
else:
resolved_inputs[k] = v
node.inputs = resolved_inputs
output = node.execute(context)
context[node_id] = output
# Convert image to base64 for return
if isinstance(output, Image.Image):
buffered = io.BytesIO()
output.save(buffered, format="PNG")
img_str = base64.b64encode(buffered.getvalue()).decode("utf-8")
results[node_id] = img_str
else:
results[node_id] = str(output)
return resultsWe will create the main FastAPI application in main.py, which serves static files and exposes a /run endpoint to execute our graph engine. This server acts as the bridge between the browser and the Python backend, handling the serialization of image data into base64 strings for easy transport over HTTP.
Next, we build a minimal static/index.html file that provides a simple interface with a 'Run Workflow' button. This frontend sends a default workflow JSON payload to the API, triggering the DAG execution and displaying the resulting images directly in the browser without requiring any heavy frontend frameworks.
main.py
import json
import base64
from io import BytesIO
from fastapi import FastAPI
from fastapi.staticfiles import StaticFiles
from fastapi.responses import HTMLResponse
from engine import execute_workflow
app = FastAPI()
# Mount the static directory to serve HTML/CSS/JS
app.mount("/static", StaticFiles(directory="static"), name="static")
@app.get("/", response_class=HTMLResponse)
def read_root():
with open("static/index.html") as f:
return f.read()
@app.post("/run")
def run_workflow(workflow: dict):
"""Execute the provided workflow graph and return base64 images."""
try:
# Execute the graph defined in the JSON payload
results = execute_workflow(workflow)
# Convert PIL Image objects to base64 strings for JSON serialization
output_images = []
for node_id, image in results.items():
buffered = BytesIO()
image.save(buffered, format="PNG")
img_str = base64.b64encode(buffered.getvalue()).decode("utf-8")
output_images.append({
"node_id": node_id,
"image_data": img_str
})
return {"status": "success", "images": output_images}
except Exception as e:
return {"status": "error", "message": str(e)}static/index.html
<!DOCTYPE html>
<html lang="en">
<head>
<meta charset="UTF-8">
<meta name="viewport" content="width=device-width, initial-scale=1.0">
<title>Node-Based Generator</title>
<style>
body { font-family: sans-serif; padding: 20px; background: #f4f4f4; }
.container { max-width: 800px; margin: 0 auto; background: white; padding: 20px; border-radius: 8px; box-shadow: 0 2px 5px rgba(0,0,0,0.1); }
button { padding: 10px 20px; background: #007bff; color: white; border: none; border-radius: 4px; cursor: pointer; font-size: 16px; }
button:hover { background: #0056b3; }
.image-grid { display: flex; gap: 10px; margin-top: 20px; flex-wrap: wrap; }
.image-card { border: 1px solid #ddd; padding: 5px; background: #fff; }
img { max-width: 100%; height: auto; display: block; }
</style>
</head>
<body>
<div class="container">
<h1>Node-Based Generator</h1>
<p>Click below to execute the default workflow graph.</p>
<button id="runBtn">Run Workflow</button>
<div id="status"></div>
<div class="image-grid" id="output"></div>
</div>
<script>
document.getElementById('runBtn').addEventListener('click', async () => {
const statusEl = document.getElementById('status');
const outputEl = document.getElementById('output');
statusEl.textContent = 'Processing...';
outputEl.innerHTML = '';
// Default workflow: Generate white image -> Convert to grayscale
const workflow = {
"nodes": [
{
"id": "gen_1",
"type": "GenerateWhiteImage",
"inputs": {},
"outputs": ["image"]
},
{
"id": "gray_1",
"type": "ConvertToGrayscale",
"inputs": { "source": "gen_1" },
"outputs": ["image"]
}
]
};
try {
const response = await fetch('/run', {
method: 'POST',
headers: { 'Content-Type': 'application/json' },
body: JSON.stringify(workflow)
});
const data = await response.json();
if (data.status === 'success') {
statusEl.textContent = 'Workflow completed!';
data.images.forEach(img => {
const card = document.createElement('div');
card.className = 'image-card';
const imgEl = document.createElement('img');
imgEl.src = `data:image/png;base64,${img.image_data}`;
imgEl.alt = `Output from ${img.node_id}`;
card.appendChild(imgEl);
const label = document.createElement('p');
label.textContent = `Node: ${img.node_id}`;
label.style.textAlign = 'center';
label.style.margin = '5px 0 0 0';
card.appendChild(label);
outputEl.appendChild(card);
});
} else {
statusEl.textContent = `Error: ${data.message}`;
}
} catch (error) {
statusEl.textContent = `Request failed: ${error.message}`;
}
});
</script>
</body>
</html>With the server and UI code in place, we start the Uvicorn development server to serve the application on localhost. This command launches the FastAPI app, making the static files and API endpoints available for immediate testing in your web browser.
Open your browser to http://localhost:8000 and click the 'Run Workflow' button. You should see two images appear: the first is a white square generated by the initial node, and the second is a grayscale version of that image, confirming that the dependency chain in the DAG executor worked correctly.
uvicorn main:app --reload --port 8000| Check | Command / Why it matters |
|---|---|
| Server starts without errors | Check terminal for 'Uvicorn running on http://127.0.0.1:8000' |
| UI loads in browser | Navigate to http://localhost:8000 and see the 'Node-Based Generator' title |
| Workflow executes | Click 'Run Workflow' and observe two images appear: one white, one grayscale |
| Dependency chain verified | Confirm the second image is visually grayscale, proving the 'ConvertToGrayscale' node received input from 'GenerateWhiteImage' |
To extend this, integrate a real Stable Diffusion backend using the diffusers library and replace the PIL dummy nodes with actual model inference calls. Consider adding a drag-and-drop library like react-flow for a full ComfyUI-like experience.