Give your AI agent an iMessage number (bring your own model)

curl -s -X PUT -H "Authorization: Bearer $ZEROSETTLE_TOKEN" -H 'Content-Type: application/json' \
  -d '{"url":"https://agent.example.com/webhook"}' https://portal.zerosettle.io/api/imessage/webhook

# agent.py: run with  uvicorn agent:app --port 8100
import asyncio, base64, hashlib, hmac, json, os, re, time
import anthropic, httpx
from fastapi import FastAPI, Request, Response

BASE = "https://portal.zerosettle.io"
TOKEN, SECRET = os.environ["ZEROSETTLE_TOKEN"], os.environ["ZEROSETTLE_WEBHOOK_SECRET"]
SYSTEM = "You are Pathfinch, an AI travel planner. Plain text; one to three short bubbles split by blank lines."
app, llm = FastAPI(), anthropic.AsyncAnthropic()        # reads ANTHROPIC_API_KEY
zs = httpx.AsyncClient(base_url=BASE, headers={"Authorization": f"Bearer {TOKEN}"}, timeout=90)
seen, queues, muted = set(), {}, set()                  # keep seen and muted in your database
VISION = ("image/jpeg", "image/png", "image/gif", "image/webp")

def verify(header: str, raw: bytes) -> bool:
    try:
        p = dict(x.split("=", 1) for x in header.split(",")); t = int(p["t"])
    except Exception:
        return False
    if abs(time.time() - t) > 300:
        return False
    exp = hmac.new(SECRET.encode(), f"{t}.".encode() + raw, hashlib.sha256).hexdigest()
    return hmac.compare_digest(exp, p["v1"])

@app.post("/webhook")
async def webhook(req: Request):
    raw = await req.body()
    if not verify(req.headers.get("X-ZeroSettle-Signature", ""), raw):
        return Response(status_code=401)
    ev = json.loads(raw)
    if ev["event"] != "message.received" or (ev["message"]["id"], ev["event"]) in seen:
        return {"ok": True}                             # message.sent echoes and retries
    m = ev["message"]; seen.add((m["id"], ev["event"]))
    if m["handle"] not in queues:
        queues[m["handle"]] = asyncio.Queue(); asyncio.create_task(worker(m["handle"]))
    queues[m["handle"]].put_nowait(m)
    return {"ok": True}

async def worker(handle: str):
    while True:
        burst = [await queues[handle].get()]
        try:                                            # wait for 3 s of quiet
            while True: burst.append(await asyncio.wait_for(queues[handle].get(), timeout=3))
        except asyncio.TimeoutError:
            pass
        try: await reply(handle, burst)
        except Exception as e: print("reply failed", handle, repr(e))

async def reply(handle: str, burst: list[dict]):
    burst.sort(key=lambda m: m["id"])                   # retries can arrive out of order
    for m in burst:                                     # STOP before any model call
        t = (m.get("text") or "").strip().lower()
        if re.search(r"\bstop\b", t): muted.add(handle)
        elif t in ("start", "unstop"): muted.discard(handle)
    if handle in muted:
        return
    typing = asyncio.create_task(keep_typing(handle))
    try:
        r = await llm.beta.messages.create(
            model="claude-opus-5-5", max_tokens=16000, output_config={"effort": "low"},
            betas=["server-side-fallback-2026-07-01"], extra_body={"fallbacks": "default"},
            system=[{"type": "text", "text": SYSTEM, "cache_control": {"type": "ephemeral"}}],
            messages=await thread(handle, burst[0]["id"], burst[-1]["id"]))
    finally:
        typing.cancel()
    text = "".join(b.text for b in r.content if b.type == "text").strip()
    if r.stop_reason == "refusal" or not text:
        await zs.post("/api/imessage/typing", json={"to": handle, "state": "stop"}); return
    for bubble in [b.strip() for b in text.split("\n\n") if b.strip()][:3]:
        await send(handle, bubble)

async def keep_typing(handle: str):
    while True:
        await zs.post("/api/imessage/typing", json={"to": handle, "state": "start", "ttl_s": 60})
        await asyncio.sleep(20)

async def thread(handle: str, first: int, last: int) -> list[dict]:
    rows = (await zs.get("/api/imessage/messages", params={
        "handle": handle, "before_id": last + 1, "limit": 40})).json()["messages"]
    out = [{"role": "user", "content": [{"type": "text", "text": "(conversation so far)"}]}]
    for r in rows:
        if r["status"] == "failed" or r["kind"] == "reaction":
            continue
        role = "user" if r["direction"] == "in" else "assistant"
        text = "[voice message]" if r["kind"] == "audio" else r["text"] or "[video or file]"
        blocks = [{"type": "text", "text": text}]
        if (r["kind"] == "image" and r.get("image") and role == "user" and r["id"] >= first
                and r["image"]["content_type"] in VISION):
            img = (await zs.get(r["image"]["url"])).content
            blocks.insert(0, {"type": "image", "source": {"type": "base64",
                "media_type": r["image"]["content_type"], "data": base64.b64encode(img).decode()}})
        if out[-1]["role"] == role: out[-1]["content"] += blocks
        else: out.append({"role": role, "content": blocks})
    return out

async def send(handle: str, text: str):
    for attempt in range(4):
        r = await zs.post("/api/imessage/send", json={"to": handle, "text": text[:4000]})
        if r.status_code == 200: return
        err = r.json().get("error", {})
        if r.status_code not in (429, 503): raise RuntimeError(f"send failed: {err.get('code')}")
        await asyncio.sleep(err.get("retry_after_s") or 2 * (attempt + 1))
    raise RuntimeError(f"send failed after retries: {err.get('code')}")