Mirror Fit SDK

A Python SDK for running ComfyUI workflows on RunPod — supports both serverless endpoints and persistent pod instances through a single unified client.

No need to match hardcoded filenames in your workflow JSON. Pass any local image paths and the SDK auto-patches the LoadImage nodes in order.

Installation

pip install mirror-fit-sdk

Quick Start

from mirror_fit import MirrorFitClient

client = MirrorFitClient()

# Serverless endpoint
sl  = client.serverless(api_key="YOUR_KEY", endpoint_id="YOUR_ENDPOINT_ID")

# Persistent pod
pod = client.pod(pod_id="YOUR_POD_ID", api_key="YOUR_KEY")

---

Serverless Usage

Use this when your ComfyUI is deployed as a RunPod serverless endpoint.

Initialize

from mirror_fit import MirrorFitClient

client = MirrorFitClient()
sl = client.serverless(
    api_key="YOUR_RUNPOD_API_KEY",
    endpoint_id="YOUR_RUNPOD_ENDPOINT_ID",
)

Check Health

health = sl.get_health()
print(health)
# {"workers": {"idle": 2, "running": 0}, "jobs": {...}}

Run Synchronously

Blocks until the generation completes and returns results.

result = sl.run_sync(
    workflow="workflow_api.json",   # path or pre-loaded dict
    images=["photo.jpg", "bg.png"], # any filenames — SDK patches the workflow
)
print(result["output"]["images"])   # list of base64 strings

Run Asynchronously

Submit a job and collect the result later — ideal for long workflows.

# Submit and return immediately
job_id = sl.run_async(workflow="workflow_api.json", images=["photo.jpg"])

# Check status without blocking
status = sl.get_status(job_id)
# {"id": "...", "status": "IN_PROGRESS"}

# Block until done and get result
result = sl.get_result(job_id)
print(result["output"]["images"])

Cancel a Job

sl.cancel(job_id)

Clear the Queue

sl.purge_queue()

---

Pod Usage

Use this when your ComfyUI is running on a persistent RunPod pod (always-on GPU).

Initialize

from mirror_fit import MirrorFitClient

client = MirrorFitClient()
pod = client.pod(
    pod_id="YOUR_POD_ID",          # "abc123xyz" — -8188 suffix handled automatically
    api_key="YOUR_RUNPOD_API_KEY", # omit if pod proxy port is public
)

Check Health

health = pod.get_health()
# {"status": "HEALTHY", "message": "ComfyUI Server is active and warming VRAM."}

List Available Models

Instantly see what checkpoints, VAEs, and LoRAs are installed on the pod — useful for debugging missing model errors.

models = pod.get_models()
print(models["checkpoints"])  # ["flux-2-klein-9b-fp8.safetensors", ...]
print(models["vaes"])         # ["flux2-vae.safetensors"]
print(models["loras"])        # [...]

Check the Queue

q = pod.get_queue()
print(f"Running: {len(q['queue_running'])}, Pending: {len(q['queue_pending'])}")

Run Synchronously

result = pod.run_sync(
    workflow="workflow_api.json",
    images=["photo.jpg", "bg.png"],  # any filenames — SDK patches the workflow
    timeout_seconds=300,             # default 240
)
print(result["output"]["images"])    # list of base64 strings

Run Asynchronously

# Submit without waiting
prompt_id = pod.run_async(workflow="workflow_api.json", images=["photo.jpg"])

# Collect when ready
result = pod.get_result(prompt_id, timeout_seconds=300)
print(result["output"]["images"])

Fetch History

# Full history of recent executions
history = pod.get_history()

# History for a specific prompt
history = pod.get_history(prompt_id="some-prompt-id")

Upload an Image

Pre-upload an image independently of running a workflow.

pod.upload_image("photos/portrait.jpg")
pod.upload_image("photos/bg.png", upload_name="background.png")

Interrupt a Running Generation

pod.interrupt()

---

Local Usage

Use this when you have a local ComfyUI running on your own machine (e.g. a local RTX 5090).

Initialize

from mirror_fit import MirrorFitClient

client = MirrorFitClient()
local = client.local(host="127.0.0.1", port=8188)

The LocalClient has the exact same API as the PodClient, so you can use get_health, get_models, run_sync, and run_async just like you would with a pod.

---

Image Handling

Both clients accept images as a list or dict. The SDK automatically patches the workflow's LoadImage nodes in order — you never have to match filenames hardcoded inside the JSON.

# List — positional: 1st image → 1st LoadImage node, 2nd → 2nd, etc.
images=["input/photo.jpg", "input/bg.png"]

# Dict — explicit upload name → local path (same positional patching applies)
images={"portrait.jpg": "input/photo.jpg", "background.png": "input/bg.png"}

---

API Reference

MirrorFitClient

Stateless factory — holds no credentials.

Method	Returns	Description
.serverless(api_key, endpoint_id)	ServerlessClient	Client for a RunPod serverless endpoint
.pod(pod_id, api_key=None)	PodClient	Client for a persistent RunPod pod

---

ServerlessClient

Obtained via client.serverless(api_key, endpoint_id).

Method	Description
get_health()	Endpoint health and worker counts
run_sync(workflow, images, timeout=300)	Execute workflow, block until done
run_async(workflow, images) → job_id	Submit job, return immediately
get_status(job_id)	Check job status without fetching output
get_result(job_id)	Poll until done, return full result
cancel(job_id)	Cancel a queued or running job
purge_queue()	Remove all pending jobs from the endpoint

run_sync / run_async parameters

Parameter	Type	Required	Description
workflow	str | dict	✅	Path to workflow_api.json or pre-loaded dict
images	list[str] | dict[str,str]	✅	Local paths (list) or {upload_name: local_path}

---

PodClient

Obtained via client.pod(pod_id, api_key=None).

Method	Description
get_health()	Check if ComfyUI server is alive
get_queue()	View running and pending prompts
get_history(prompt_id=None)	Retrieve past execution history
get_models()	List installed checkpoints, VAEs, LoRAs, CLIPs, UNETs
upload_image(filepath, upload_name=None)	Upload an image to the pod's input folder
interrupt()	Stop the currently running generation
run_sync(workflow, images, timeout_seconds=240)	Execute workflow, block until done
run_async(workflow, images) → prompt_id	Queue prompt, return immediately
get_result(prompt_id, timeout_seconds=240)	Poll until done, return full result

run_sync / run_async parameters

Parameter	Type	Required	Description
workflow	str | dict	✅	Path to workflow_api.json or pre-loaded dict
images	list[str] | dict[str,str]	✅	Local paths (list) or {upload_name: local_path}
timeout_seconds	int	❌	Max wait before TimeoutError (default 240)

---

Response Format

All run_sync and get_result calls return the same shape:

{
  "id": "prompt-uuid",
  "status": "COMPLETED",
  "delayTime": 1200,
  "executionTime": 8500,
  "output": {
    "message": "Generation completed successfully.",
    "images": ["<base64>", "<base64>"]
  }
}

---

Saving Output Images

import base64, os

def save_images(result, output_dir="output"):
    os.makedirs(output_dir, exist_ok=True)
    for i, b64 in enumerate(result["output"]["images"]):
        path = os.path.join(output_dir, f"output_{i+1}.png")
        with open(path, "wb") as f:
            f.write(base64.b64decode(b64))
        print(f"Saved: {path}")

result = pod.run_sync(workflow="workflow_api.json", images=["photo.jpg"])
save_images(result)

---

Error Handling

import requests

try:
    result = pod.run_sync(workflow="workflow_api.json", images=["photo.jpg"])
except TimeoutError:
    print("Workflow timed out — try increasing timeout_seconds")
except requests.HTTPError as e:
    print(f"HTTP {e.response.status_code}: {e}")   # includes ComfyUI error body for 500s
except RuntimeError as e:
    print(f"Job failed: {e}")                       # raised by get_result on FAILED jobs
except Exception as e:
    print(f"Unexpected error: {e}")

---

Requirements

• Python 3.9+
• requests >= 2.32.0
• Active RunPod account
• A deployed ComfyUI serverless endpoint and/or a running ComfyUI pod

License

MIT