mirror-fit-sdk 0.1.0

Python SDK for running ComfyUI workflows on RunPod — serverless endpoints and persistent pods.

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.

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)

Example response:

{ "workers": { "idle": 2, "running": 0 } }

Run a Workflow

Pass a path to a workflow_api.json file (ComfyUI API format) and any input images:

# Images as a list — filenames are used as ComfyUI node identifiers
result = sl.run_sync(
    workflow="workflow_api.json",
    images=["image1.jpg", "image2.jpg"],
)

# Images as a dict — map custom ComfyUI node names to local paths
result = sl.run_sync(
    workflow="workflow_api.json",
    images={"my_input.jpg": "path/to/local/image.jpg"},
)

Full Serverless Example

from mirror_fit import MirrorFitClient

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

print("Health:", sl.get_health())

try:
    result = sl.run_sync(
        workflow="workflow_api.json",
        images=["image1.jpg"],
    )
    print("Output images (base64):", result["output"]["images"])
except Exception as e:
    print(f"Error: {e}")

---

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",          # e.g. "abc123xyz" — the -8188 suffix is handled automatically
    api_key="YOUR_RUNPOD_API_KEY", # optional if the pod proxy port is public
)

Check Health

health = pod.get_health()
print(health)

Example response:

{ "status": "HEALTHY", "message": "ComfyUI Server is active and warming VRAM." }

Run a Workflow

# Images as a list
result = pod.run_sync(
    workflow="workflow_api.json",
    images=["image1.jpg", "image2.jpg"],
    timeout_seconds=300,  # optional, default 240
)

# Images as a dict
result = pod.run_sync(
    workflow="workflow_api.json",
    images={"my_input.jpg": "path/to/local/image.jpg"},
)

Full Pod Example

from mirror_fit import MirrorFitClient

client = MirrorFitClient()
pod = client.pod(pod_id="abc123xyz", api_key="YOUR_RUNPOD_API_KEY")

print("Health:", pod.get_health())

try:
    result = pod.run_sync(
        workflow="workflow_api.json",
        images=["image1.jpg"],
        timeout_seconds=300,
    )
    print("Output images (base64):", result["output"]["images"])
except TimeoutError:
    print("Workflow timed out.")
except Exception as e:
    print(f"Error: {e}")

---

API Reference

MirrorFitClient

A stateless factory. No credentials are stored on this object.

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

---

ServerlessClient

Obtained via client.serverless(...).

Constructor

Parameter	Type	Required	Description
api_key	str	✅	Your RunPod API key
endpoint_id	str	✅	Your RunPod serverless endpoint ID

Methods

get_health() → dict Check the health status of the serverless endpoint.

run_sync(workflow, images) → dict

Parameter	Type	Required	Description
workflow	str | dict	✅	Path to workflow_api.json or a pre-loaded dict
images	list[str] | dict[str, str]	✅	List of local paths or {node_name: local_path} dict

---

PodClient

Obtained via client.pod(...).

Constructor

Parameter	Type	Required	Description
pod_id	str	✅	Your RunPod pod ID (e.g. "abc123xyz")
api_key	str	❌	API key — only needed for private proxy ports

Methods

get_health() → dict Check whether the ComfyUI server on the pod is alive.

run_sync(workflow, images, timeout_seconds=240) → dict

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

---

Response Format

Both ServerlessClient.run_sync() and PodClient.run_sync() return a compatible response shape:

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

---

Error Handling

try:
    result = sl.run_sync(workflow="workflow_api.json", images=["input.jpg"])
except TimeoutError:
    print("Execution timed out.")
except requests.HTTPError as e:
    print(f"HTTP error: {e.response.status_code}")
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