Twin API

# 1. Create the twin. Returns 202 with status: queued.
curl https://api.metaphi.ai/v1/kennet/twins \
  -H "Authorization: Bearer $API_KEY" \
  -H "Idempotency-Key: $(uuidgen)" \
  -F file=@trace.jsonl
# {"id": "twin_4FFTGxX_b40dXo8F-ObejckZj1DsFKPhqvJFo25CNH0", "object": "kennet.twin", "status": "queued", ...}

# 2. Retrieve it until status is completed, or use a webhook.
curl https://api.metaphi.ai/v1/kennet/twins/twin_4FFTGxX_b40dXo8F-ObejckZj1DsFKPhqvJFo25CNH0 \
  -H "Authorization: Bearer $API_KEY"

# 3. Download the twin.
curl https://api.metaphi.ai/v1/kennet/twins/twin_4FFTGxX_b40dXo8F-ObejckZj1DsFKPhqvJFo25CNH0/content \
  -H "Authorization: Bearer $API_KEY" -o trace.twin.jsonl

import os, time, uuid
import requests

API = "https://api.metaphi.ai/v1/kennet"
s = requests.Session()
s.headers["Authorization"] = f"Bearer {os.environ['API_KEY']}"

# 1. Upload the trace
with open("trace.jsonl", "rb") as f:
    r = s.post(f"{API}/twins", files={"file": ("trace.jsonl", f)},
               headers={"Idempotency-Key": str(uuid.uuid4())})
r.raise_for_status()
twin = r.json()

# 2. Wait for it to end
delay = 5
while twin["status"] not in ("completed", "failed"):
    time.sleep(delay)
    delay = min(delay * 2, 60)
    twin = s.get(f"{API}/twins/{twin['id']}").json()
if twin["status"] == "failed":
    raise SystemExit(twin["error"]["message"])

# 3. Download the twin
with s.get(f"{API}/twins/{twin['id']}/content", stream=True) as r:
    r.raise_for_status()
    with open("trace.twin.jsonl", "wb") as out:
        for chunk in r.iter_content(1 << 20):
            out.write(chunk)