Automating Workflows with the ComfyUI API

ComfyUI's HTTP API opens up powerful possibilities for integrating image generation workflows into external applications, automation platforms, and custom solutions. This lesson explores how to leverage the API for programmatic control and workflow automation.

Understanding the ComfyUI HTTP API

The ComfyUI HTTP API provides RESTful endpoints that allow you to:

• Submit workflow definitions programmatically
• Monitor execution status
• Retrieve generated results
• Manage queue operations
• Access system information

Key API Endpoints

The primary endpoints include:

• POST /prompt - Submit a workflow for execution
• GET /history - Retrieve execution history
• GET /queue - Check current queue status
• WebSocket /ws - Real-time status updates
• GET /object_info - Node and parameter information

Setting Up API Integration

To begin working with the ComfyUI API, ensure your ComfyUI instance is running with API access enabled:

import requests
import json
import websocket
import uuid

# ComfyUI server configuration
SERVER_ADDRESS = "127.0.0.1:8188"
CLIENT_ID = str(uuid.uuid4())

def queue_prompt(prompt):
    """Submit a workflow to ComfyUI for execution"""
    p = {"prompt": prompt, "client_id": CLIENT_ID}
    data = json.dumps(p).encode('utf-8')
    req = requests.post(f"http://{SERVER_ADDRESS}/prompt", data=data)
    return req.json()

def get_image(filename, subfolder, folder_type):
    """Retrieve generated image from ComfyUI"""
    data = {"filename": filename, "subfolder": subfolder, "type": folder_type}
    url_values = urllib.parse.urlencode(data)
    return requests.get(f"http://{SERVER_ADDRESS}/view?{url_values}").content

WebSocket Integration for Real-Time Updates

For monitoring workflow execution in real-time, implement WebSocket communication:

import websocket
import json

def on_message(ws, message):
    """Handle WebSocket messages from ComfyUI"""
    if isinstance(message, str):
        message = json.loads(message)
        if message['type'] == 'status':
            data = message['data']
            if data['status']['exec_info']['queue_remaining'] == 0:
                print("Workflow completed!")
        elif message['type'] == 'progress':
            print(f"Progress: {message['data']['value']}/{message['data']['max']}")
        elif message['type'] == 'executed':
            output = message['data']['output']
            print(f"Node executed with output: {output}")

# Connect to WebSocket
ws = websocket.WebSocketApp(
    f"ws://{SERVER_ADDRESS}/ws?clientId={CLIENT_ID}",
    on_message=on_message
)
ws.run_forever()

Integration with External Automation Tools

n8n Integration Example

ComfyUI can be integrated with workflow automation platforms like n8n:

// n8n HTTP Request node configuration
const workflowData = {
  "prompt": {
    "1": {
      "inputs": {
        "text": "{{$json.prompt}}",
        "clip": ["4", 1]
      },
      "class_type": "CLIPTextEncode"
    },
    "2": {
      "inputs": {
        "samples": ["3", 0],
        "vae": ["4", 2]
      },
      "class_type": "VAEDecode"
    }
  },
  "client_id": "n8n-automation"
};

return {
  json: workflowData
};

Python Automation Script

Here's a complete automation script that demonstrates workflow submission and result retrieval:

import requests
import json
import time
import base64
from io import BytesIO
from PIL import Image

class ComfyUIAutomator:
    def __init__(self, server_address="127.0.0.1:8188"):
        self.server_address = server_address
        self.client_id = str(uuid.uuid4())
    
    def submit_workflow(self, workflow_json, prompt_text):
        """Submit a parameterized workflow"""
        # Modify workflow with dynamic parameters
        workflow_json["1"]["inputs"]["text"] = prompt_text
        
        prompt_data = {
            "prompt": workflow_json,
            "client_id": self.client_id
        }
        
        response = requests.post(
            f"http://{self.server_address}/prompt",
            json=prompt_data
        )
        
        if response.status_code == 200:
            return response.json()["prompt_id"]
        else:
            raise Exception(f"Failed to submit workflow: {response.text}")
    
    def wait_for_completion(self, prompt_id, timeout=300):
        """Wait for workflow completion"""
        start_time = time.time()
        
        while time.time() - start_time < timeout:
            history = requests.get(f"http://{self.server_address}/history/{prompt_id}")
            
            if history.status_code == 200:
                history_data = history.json()
                if prompt_id in history_data:
                    return history_data[prompt_id]
            
            time.sleep(2)
        
        raise TimeoutError("Workflow execution timed out")
    
    def download_results(self, history_data):
        """Download generated images"""
        outputs = history_data.get("outputs", {})
        images = []
        
        for node_id, node_output in outputs.items():
            if "images" in node_output:
                for image_data in node_output["images"]:
                    filename = image_data["filename"]
                    subfolder = image_data.get("subfolder", "")
                    
                    image_url = f"http://{self.server_address}/view"
                    params = {
                        "filename": filename,
                        "subfolder": subfolder,
                        "type": "output"
                    }
                    
                    response = requests.get(image_url, params=params)
                    if response.status_code == 200:
                        images.append(Image.open(BytesIO(response.content)))
        
        return images

# Usage example
automator = ComfyUIAutomator()
workflow = load_workflow_json("my_workflow.json")

prompt_id = automator.submit_workflow(workflow, "a beautiful sunset landscape")
result = automator.wait_for_completion(prompt_id)
images = automator.download_results(result)

for i, img in enumerate(images):
    img.save(f"generated_image_{i}.png")

Scaling and Production Considerations

For production deployments, consider:

1. Load Balancing: Distribute requests across multiple ComfyUI instances
2. Queue Management: Implement proper queue monitoring and error handling
3. Resource Optimization: Monitor GPU usage and implement auto-scaling
4. Error Recovery: Handle network failures and workflow errors gracefully
5. Security: Implement authentication and rate limiting for public APIs

Best Practices

• Always validate workflow JSON before submission
• Implement proper error handling and retry logic
• Use WebSocket connections for real-time monitoring
• Cache frequently used models and workflows
• Monitor system resources and implement appropriate scaling
• Log all API interactions for debugging and monitoring