Python SDK

modelreins-worker is the official Python SDK for building silicon workers. It’s stdlib-only — no requests, no httpx, no dependencies — and the whole public API fits on one screen.

Published on PyPI: pypi.org/project/modelreins-worker/

Install

pip install modelreins-worker

If pip complains about PEP 668 (externally-managed environment, common on newer Ubuntu and Debian), use a venv:

python3 -m venv mr-worker && source mr-worker/bin/activate
pip install modelreins-worker

60-second quickstart

1. Register a worker

On the Workers dashboard, click Register, name it, pick capabilities, get a one-time-view URL. Open the URL once, copy the raw token, save it.

2. Write the worker

from modelreins_worker import Worker

def handle_job(job):
    prompt = job["prompt"]
    # ... do the work ...
    return "Done"

Worker(name="my-first-worker").run(handler=handle_job)

3. Run it

export MODELREINS_TOKEN=<your-worker-token>
export MODELREINS_URL=https://app.modelreins.com   # optional; this is the default
python my_worker.py

The worker heartbeats every 5 seconds, polls its inbox, claims pending jobs assigned to its name, and calls your handle_job for each one. Exceptions auto-fail the job with the traceback as output.

API surface

class Worker:
    def __init__(self,
                 token=None,         # or MODELREINS_TOKEN env var
                 url=None,           # or MODELREINS_URL env var; default app.modelreins.com
                 name=None,          # defaults to hostname-pid
                 poll_interval=5.0,
                 timeout=30.0): ...

    def heartbeat(self, status="active", details=None, tags=None) -> dict: ...
    def inbox(self, status="pending") -> list: ...
    def claim(self, job_id: int) -> dict: ...
    def complete(self, job_id: int, result: str = "", status: str = "done") -> dict: ...

    # New in 0.2.0 — requires server ≥ 4.6.0
    def upload_artifact(self, data: bytes, content_type: str,
                        filename=None, job_id=None, review_id=None,
                        expires_at=None) -> dict: ...
    def submit_review(self, type: str, title: str, content: str,
                      preview=None, target=None, job_id=None) -> dict: ...

    def run(self, handler, on_error=None) -> None:
        """Blocking loop: heartbeat → claim → handler(job) → complete."""


class WorkerError(Exception):
    """Raised for any non-2xx response from the ModelReins API."""

Seven methods plus one exception.

Artifacts and reviews (new in 0.2.0)

upload_artifact(data, content_type, ...)

Upload a binary blob (image, video, audio, PDF) to the modelreins artifact tunnel and get back a hosted URL.

with open("poster.png", "rb") as f:
    png = f.read()

result = worker.upload_artifact(
    data=png,
    content_type="image/png",
    filename="q4-poster.png",
)
print(result)
# {'success': True,
#  'slug': '8e165514961c4b658c771d4a277d167bb3402cef',
#  'url': '/s/8e165514961c4b658c771d4a277d167bb3402cef',
#  'content_type': 'image/png',
#  'size_bytes': 1234,
#  'filename': 'q4-poster.png'}

The returned url is the path component — prepend worker.url for the absolute URL. Or just hand the slug to whoever’s consuming it; the canonical fetch is https://app.modelreins.com/s/<slug>.

Optional kwargs:

Arg	Purpose
filename	Original filename (defaults to artifact.<ext> based on content_type)
job_id	Link the artifact back to the producing job
review_id	Link the artifact to a specific review queue item
expires_at	ISO timestamp; null = never expires

Multipart-encoded manually, no requests / httpx dependency.

submit_review(type, title, content, ...)

Push content into the human-review queue. Used when a worker produces output that shouldn’t publish without a human eyeball — typical for risk_tier=audit or risk_tier=approve capabilities.

review = worker.submit_review(
    type="generated_image",
    title="Q4 launch poster — draft 1",
    content="Auto-generated by image-gen worker at 2026-04-22 14:30",
    preview="https://app.modelreins.com/s/8e165514961c4b658c771d4a277d167bb3402cef",
    target="marketing-channel",
)
print(review)
# {'id': 268, 'status': 'pending', 'mode': 'review'}

The review item lands in /review/queue where a human operator can preview the artifact inline (via the preview URL) and Approve / Reject. Approved items publish to the target channel.

Optional job_id links the review back to the producing job.

Putting them together — image generation worker

import os
from modelreins_worker import Worker

# Pseudo-call to a text-to-image API (1minai, replicate, openai, whatever)
def generate_image(prompt: str) -> bytes:
    ...  # returns raw PNG bytes
    return png_bytes

def handle(job):
    png = generate_image(job["prompt"])
    art = worker.upload_artifact(
        data=png,
        content_type="image/png",
        filename=f"job-{job['id']}.png",
        job_id=job["id"],
    )
    worker.submit_review(
        type="generated_image",
        title=f"Job {job['id']}: {job['prompt'][:60]}",
        content=f"Generated by {worker.name}",
        preview=f"{worker.url}{art['url']}",
        job_id=job["id"],
    )
    return f"Uploaded artifact {art['slug']}, queued for review"

worker = Worker(name="image-worker-01")
worker.run(handler=handle)

A complete working version ships in the SDK source at examples/image_generation_worker.py using 1minai’s IMAGE_GENERATOR.

Handler contract

handler(job: dict) -> str — your function receives the job dict (contains at least id, prompt, assigned_to, status) and returns a string that becomes the job’s output. Any exception auto-fails the job and writes the traceback as the error.

If you need different behavior on errors, pass on_error=your_handler to run():

def on_error(job, exception):
    notify_pagerduty(job["id"], exception)

Worker(name="my-worker").run(handler=handle_job, on_error=on_error)

Authentication

The SDK uses worker tokens — project-scoped credentials minted through the workers dashboard with declared capabilities. Worker tokens authenticate as the worker identity; every action they take is attributed to that worker in your audit log.

Worker tokens are distinct from user API keys:

• User API keys mint at /settings/api-keys and authenticate as you.
• Worker tokens mint at /workers and authenticate as a specific registered bot.

Both are delivered through the same one-time-view URL flow — raw values exist in our database for minutes, then self-delete on first view or expiry.

Environment variables

Variable	Default	What it does
MODELREINS_TOKEN	(required)	Worker token from registration
MODELREINS_URL	https://app.modelreins.com	API base

Version compatibility

modelreins-worker tracks its own SemVer independent of the ModelReins server version. The 0.x track signals that the SDK’s Python API may evolve — going 1.0 is a commitment to not break Worker() users.

Compatible with ModelReins server ≥ 4.5.0 for the base API. The upload_artifact() and submit_review() methods (added in SDK 0.2.0) require server ≥ 4.6.0.

Source and issues

Published on PyPI: modelreins-worker. For issues or feature requests, email [email protected].