modelreins-worker

The Python SDK for building silicon workers on ModelReins.

ModelReins is an AI workforce dispatcher. A silicon worker is a bot that registers with capabilities, heartbeats its presence, and runs dispatched jobs — treated as a first-class employee in your tenant, not an anonymous API key.

stdlib-only. No requests, no httpx — just urllib. ~150 LOC.

Install

pip install modelreins-worker

60-second quickstart

1. Register your worker

Mint a worker token (admin or master-token required, one-time-view URL returned):

curl -X POST https://app.modelreins.com/api/v1/workers/register \
  -H "Authorization: Bearer $MODELREINS_ADMIN_TOKEN" \
  -H "Content-Type: application/json" \
  -d '{
    "name": "my-first-worker",
    "description": "Handles triage on new support emails",
    "capabilities": [
      {"name": "email.triage", "risk_tier": "auto"},
      {"name": "email.reply",  "risk_tier": "audit"}
    ]
  }'

Response includes view_url — open it once to reveal the raw token. Save it as MODELREINS_TOKEN.

2. Write your worker

from modelreins_worker import Worker

def handle_job(job):
    # job is a dict with at least {id, prompt, assigned_to, ...}
    prompt = job["prompt"]
    # ... do the work ...
    return "Triage complete: priority=high, category=billing"

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 will heartbeat every 5s, poll /my-jobs?worker=my-first-worker for pending assignments, and call your handle_job for each one. Exceptions in handle_job auto-mark the job failed with the traceback as output.

API surface

class Worker:
    def __init__(self, token=None, url=None, name=None,
                 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: ...

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

All methods raise WorkerError on non-2xx responses.

Capabilities + risk tiers

When you register a worker, you declare its capabilities — a list of {name, risk_tier, requires_secrets?} entries.

Risk tiers:

requires_secrets (optional) is a list of vault://... refs that the Director will resolve from your vault at dispatch time. ModelReins itself holds no credential material — see /settings/vault on your tenant dashboard.

Zero-data stance

ModelReins is a dispatch relay, not a content host. Your workers' prompts and outputs stay in your tenant. The server has routing metadata (who, when, which capability, which status). It does not — and at the policy level will not — query your content across tenants. See the threat model at https://modelreins.com/security (coming soon).

License

MIT.