warpswarm (Python)

On-demand human judgment for AI evaluation and labeling. Submit items, get aggregated human answers back.

pip install warpswarm
export WARPSWARM_API_KEY=ws_live_...     # ws_test_... = free sandbox, no charges

import warpswarm as ws

client = ws.Client()

job = client.compare(
    "Summary A/B",
    "Which summary is more accurate?",
    [(summary_a[i], summary_b[i]) for i in range(100)],   # (a, b) pairs: text, Path, URL or asset id
    answers=5,                                           # answers per item
)                                                         # budget defaults to items × answers × the job's price
job.add_gold([(("Paris is in France.", "Paris is in Spain."), "a")])  # optional known-answer checks
job.start().wait(on_progress=print)

for r in job.results():
    print(r.external_id, r.aggregate.winner, r.aggregate.votes, r.confidence)
job.export("results.csv", format="csv")

Task types

Coordinates are fractions of the image (0–1), x, y = top-left. Media can be a string (text), a pathlib.Path or bytes (uploaded once), an http(s) URL (imported), an ast_… id, or an Asset. Any item can also be a full dict with context (e.g. the prompt) and external_id (your own id, returned on every result).

Gold correct values: compare "a"|"b"|"tie" · classify "label" or [labels] · free_text "text" · locate a region (x, y, w, h) (any tap inside passes) · draw [(x, y, w, h), …] · select_words [index] or ["word", …] · ranking [index, best first].

Lifecycle and pricing

draft → start() → queued → running → completed. start() holds credit for the job's budget; unused credit is released at the end, and only accepted answers are charged.

• Price per 1,000 answers. $10 base; $13 compare with images; $18 any job with video; $50 free_text. client.estimate(task_type, n, answers, media=["video"]) returns it; job.price_per_answer_micros shows it.

• Phases. job.progress.phase is waiting_for_workers → in_review → collecting → done. wait() polls at most once a minute before workers start and every poll_interval s while collecting.

• Minimum size. A job needs at least 14 items (ws.MIN_DATAPOINTS); smaller jobs raise JobTooSmallError.

• Unsold tail. Workers answer items in batches. If fewer than 14 answers remain at the end, they can't be staffed: the job completes, progress.answers_unsold says how many, and they are never charged.

Webhooks

Pass webhook_url="https://…" when creating a job to get job.started, job.progress, job.completed, job.cancelled and job.failed POSTs instead of polling. job.webhook_secret is returned once, at creation, so store it.

event = ws.verify_webhook(request.body, request.headers["Warpswarm-Signature"], secret)   # raw body bytes
if event.type == "job.completed":
    job = client.job(event.job_id)

verify_webhook checks the HMAC and rejects timestamps older than 5 minutes (ws.WebhookSignatureError). Retries reuse event.id, so dedupe on it. Any 2xx acknowledges; failures are retried for 24 h.

Errors

All errors subclass ws.WarpswarmError (.status, .code, .param): AuthenticationError, NotFoundError, InvalidRequestError, JobTooSmallError, InvalidStateError, InsufficientCreditError, RateLimitError, ServerError, plus WaitTimeout / JobFailed from wait(). Every POST carries an Idempotency-Key, so the SDK retries network errors, 429 and 5xx safely.

Other calls

client.account(), client.estimate(task_type, n, answers), client.upload(path), client.upload_url(url), client.jobs(status=), client.job(id), job.preview_url() (see the exact worker form), job.pause()/resume()/cancel(), job.results(include_answers=True) (each answer with a pseudonymous annotator), client.checkout_url(amount_usd).

Development

uv venv .venv && uv pip install -e . pytest pyyaml openapi-schema-validator
pytest tests/test_unit.py            # offline; every request body is validated against ../../api/openapi.yaml
# full loop, all 7 types (local wrangler dev + worker/tests/mock/mw_mock.py on :8799):
WS_BASE=http://localhost:8787 WS_ADMIN=… WS_SEED="npx wrangler d1 execute warpswarm --local --command" pytest tests/test_integration.py
# prod, read + draft lifecycle only (no workers, no charges):
WS_BASE=https://api.warpswarm.app WS_TEST_KEY=ws_test_… pytest tests/test_integration.py -k prod
uv build                             # dist/warpswarm-0.1.0-py3-none-any.whl