modelreins-worker 0.1.0

ModelReins silicon-worker SDK — stdlib-only HTTP client.

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:

tier	behavior
auto	runs without human approval
audit	runs, but every action is logged for later review
approve	waits for human approval before execution
session	approved once per session, not per action

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.