Metadata-Version: 2.4
Name: mirror-fit-sdk
Version: 0.2.2
Summary: Python SDK for running ComfyUI workflows on RunPod — serverless endpoints and persistent pods.
Project-URL: Homepage, https://github.com/tritosmim/RunPodComfySDK
Project-URL: Repository, https://github.com/tritosmim/RunPodComfySDK
Project-URL: Bug Tracker, https://github.com/tritosmim/RunPodComfySDK/issues
Author-email: Tosmim <tosmim.mehtab@trinetralabs.ai>
License: MIT License
        
        Copyright (c) 2026 Tosmim
        
        Permission is hereby granted, free of charge, to any person obtaining a copy
        of this software and associated documentation files (the "Software"), to deal
        in the Software without restriction, including without limitation the rights
        to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
        copies of the Software, and to permit persons to whom the Software is
        furnished to do so, subject to the following conditions:
        
        The above copyright notice and this permission notice shall be included in all
        copies or substantial portions of the Software.
        
        THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
        IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
        FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
        AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
        LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
        OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
        SOFTWARE.
License-File: LICENSE
Keywords: ai,comfyui,image-generation,mirror-fit,runpod,sdk,stable-diffusion
Classifier: Development Status :: 3 - Alpha
Classifier: Intended Audience :: Developers
Classifier: License :: OSI Approved :: MIT License
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.9
Classifier: Programming Language :: Python :: 3.10
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Topic :: Scientific/Engineering :: Artificial Intelligence
Classifier: Topic :: Software Development :: Libraries :: Python Modules
Requires-Python: >=3.9
Requires-Dist: requests>=2.32.0
Description-Content-Type: text/markdown

# 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

```bash
pip install mirror-fit-sdk
```

## Quick Start

```python
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

```python
from mirror_fit import MirrorFitClient

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

### Check Health

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

### Run Synchronously

Blocks until the generation completes and returns results.

```python
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.

```python
# 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

```python
sl.cancel(job_id)
```

### Clear the Queue

```python
sl.purge_queue()
```

---

## Pod Usage

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

### Initialize

```python
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

```python
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.

```python
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

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

### Run Synchronously

```python
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

```python
# 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

```python
# 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.

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

### Interrupt a Running Generation

```python
pod.interrupt()
```

---

## 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.

```python
# 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:

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

---

## Saving Output Images

```python
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

```python
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](https://runpod.io) account
- A deployed ComfyUI **serverless endpoint** and/or a running ComfyUI **pod**

## License

MIT
