Skip to content

Client / integration

For applications calling a hosted Mantis instance.

Topic What it covers
Authentication Getting a token, request headers, scope, error responses
Sending plans Request body shape for each plan format, run options, caps
Runs and polling Detached runs, status/result/logs/cancel actions, idempotency, webhooks
Recordings Requesting screencasts, downloading polished or raw, action overlays
Errors Error codes, retry guidance, debugging

Looking for the full HTTP API surface? See Reference / HTTP API.

Minimal client (Python)

import os, time, requests

ENDPOINT = os.environ["MANTIS_ENDPOINT"]      # e.g. https://model-qvvgkneq.api.baseten.co/production/sync
PLATFORM = os.environ["BASETEN_API_KEY"]
TENANT   = os.environ["MANTIS_API_TOKEN"]

def headers():
    h = {"X-Mantis-Token": TENANT, "Content-Type": "application/json"}
    if PLATFORM:
        h["Authorization"] = f"Api-Key {PLATFORM}"
    return h

def submit(plan_path: str, state_key: str, **opts) -> str:
    r = requests.post(f"{ENDPOINT}/v1/predict", headers=headers(), json={
        "detached": True,
        "micro": plan_path,
        "state_key": state_key,
        **opts,
    }, timeout=60)
    r.raise_for_status()
    return r.json()["run_id"]

def poll(run_id: str, every: int = 30) -> dict:
    while True:
        r = requests.post(f"{ENDPOINT}/v1/predict", headers=headers(), json={
            "action": "status", "run_id": run_id,
        }, timeout=60)
        r.raise_for_status()
        body = r.json()
        if body["status"] in {"succeeded", "failed", "cancelled"}:
            return body
        time.sleep(every)

def fetch_result(run_id: str) -> dict:
    r = requests.post(f"{ENDPOINT}/v1/predict", headers=headers(), json={
        "action": "result", "run_id": run_id,
    }, timeout=60)
    r.raise_for_status()
    return r.json()

# Usage
run_id = submit(
    "plans/example/extract_listings.json",
    "marketplace-prod",
    max_cost=2, max_time_minutes=20, record_video=True,
)
final = poll(run_id)
print(final["summary"])
result = fetch_result(run_id)
print(result["result"]["leads"])

That's the whole integration. The rest of this section is the deep-dive on each piece.