Sync conversations into your warehouse

Support volume is only useful next to everything else you measure. This page builds the job that gets it there: a scheduled process that pulls conversations and their transcripts into your own store, picks up where it left off, survives being run twice, and does not spend your whole rate-limit budget doing it.

The API gives you three read endpoints and a cursor. Everything hard about this job is in four decisions — what you persist between runs, when to stop walking, how fast to walk, and what you do about the events that arrive while you are not looking. This page makes all four explicit and then gives you the finished job.

What you need before you start #

• An API key with conversations:read. Read conversations is ticked by default when you create a key, along with Read agents. A key's scopes can't be changed afterwards. See Scopes.
• Somewhere durable to keep a watermark. A row in a table, a key in Redis, a small JSON file next to the job. It must survive a deploy.
• An upsert, not an insert. Both conversations and messages carry stable ids. Every write in this job is keyed on one.
• A schedule. Every 5 to 15 minutes is normal. Read the pacing section below before you go faster.

The shape of the job #

1. Read the watermark you saved last time.
2. Walk GET /conversations from page one, newest activity first.
3. For every conversation whose updated_at is above the watermark, upsert it and walk its transcript.
4. Stop the moment you reach a conversation at or below the watermark, or when a short page ends the list.
5. Write the highest updated_at you processed back to the watermark — once, at the end.

Two nested loops. The outer one walks the conversation list newest-activity-first and stops as soon as it reaches something you already have. The inner one walks the transcript of every conversation the outer loop accepted. Around both sits a pacer, and above both sits the watermark.

The cursor contract #

Both paginated endpoints return the same envelope:

{ "items": [ … ], "next_cursor": "…" }

One rule governs the loop, and getting it wrong is the most common bug in this integration:

The practical consequences:

• Loop on next_cursor, never on items.length. A page of 100 followed by an empty page with next_cursor: null is normal and correct. Code that stops when it sees a full page is wrong in the other direction — it silently truncates.
• Never construct or parse a cursor. For conversations it happens to be an updated_at string; for messages it happens to be a message_id. Both are implementation details you should not be reading. Pass back what you were given, byte for byte.
• There is no total, no has_more, no page number, and no way to go backwards. If your UI needs a count, count what you stored.
• An empty first page means the workspace has no conversations. It does not mean an error. A bad conversation_id on the messages endpoint is a 404, never an empty page — so an empty transcript genuinely means an empty transcript.

Pagination is the full contract.

What you persist between runs: a watermark, not a cursor #

The instinct is to save next_cursor at the end of a run and resume from it next time. Do not. A cursor is a position inside one walk of a list that is reordering itself continuously — conversations come back ordered by updated_at descending, so every new message moves a conversation to the front and shifts everything behind it. A cursor from twenty minutes ago points at a place in a list that no longer exists.

What you persist is a watermark: the highest updated_at you have successfully processed. On the next run you walk from the top of the list again and stop the moment you reach a conversation at or below it.

This inverts the usual failure mode in a helpful direction. Consider a conversation that receives a new message while your walk is in progress:

1. It jumps to the front of the list — behind the cursor you already passed.
2. This run misses it.
3. Its updated_at is now later than anything this run observed, so it is above the watermark you save.
4. The next run sees it on page one.

The window is bounded by your schedule, not by luck. That is the whole reason to prefer a watermark.

On the very first run there is no watermark. That run is a full backfill: it walks until a short page ends it. Plan for it — the first run is covered in its own section below.

Pacing against the rate limit #

The limit is 120 requests per minute, per key, counted in a fixed one-minute window rather than a sliding one. This job is request-hungry by construction: one request per page of conversations, plus at least one per conversation that changed.

A workspace with 400 conversations updated since the last run costs roughly 4 list requests plus 400 transcript requests. At full speed that is a guaranteed 429 and, worse, it starves whatever else that key is doing.

Three things to build in:

• Pace deliberately. Space requests so you land near 90 per minute rather than 120. The headroom is for the message you need to send while the sync is running.
• Read X-RateLimit-Remaining. It is set on every successful response, alongside X-RateLimit-Limit. When remaining drops into single digits, slow down before the server has to tell you.
• Sleep to the window boundary on a 429. There is no Retry-After header and no X-RateLimit-Reset. The window is a wall-clock minute, so sleeping until the next minute boundary is exactly right and anything shorter is a wasted request.

The complete job #

Both versions below are runnable. The only things left as stubs are the four functions that talk to your own store: load_state, save_state, upsert_conversation, upsert_messages.

# sync_conversations.py
import json
import os
import time
from datetime import datetime, timedelta

import requests

BASE_URL = "https://services.amomic.in/api/v1"
API_KEY = os.environ["AMOMIC_API_KEY"]
HEADERS = {"Authorization": f"Bearer {API_KEY}"}

# 120 requests/minute is the ceiling. Aim at 90 and keep the rest for
# whatever else this key does while the sync is running.
MIN_INTERVAL_S = 60 / 90
OVERLAP = timedelta(minutes=5)
MAX_ATTEMPTS = 5
BACKOFF_S = [1, 4, 12, 30]


class SyncError(Exception):
    """A non-retryable API error. The envelope is attached for logging."""

    def __init__(self, status: int, error: dict):
        super().__init__(f"{error['code']}: {error['message']}")
        self.status = status
        self.code = error["code"]
        self.details = error.get("details") or {}
        self.reason = self.details.get("reason")
        self.request_id = error.get("request_id")


class Pacer:
    """Spaces requests, and slows down when the server says headroom is low."""

    def __init__(self) -> None:
        self._next_at = 0.0

    def wait(self) -> None:
        delay = self._next_at - time.monotonic()
        if delay > 0:
            time.sleep(delay)

    def observe(self, response: requests.Response) -> None:
        interval = MIN_INTERVAL_S
        remaining = response.headers.get("X-RateLimit-Remaining")
        # The headers are present on success only; absent on any error.
        if remaining is not None and remaining.isdigit() and int(remaining) < 20:
            interval = 1.5
        self._next_at = time.monotonic() + interval


def _sleep_to_minute_boundary() -> None:
    # Fixed one-minute window: the counter resets on the wall-clock minute.
    time.sleep(60 - (time.time() % 60) + 0.25)


def get(client: requests.Session, path: str, params: dict, pacer: Pacer) -> dict:
    for attempt in range(MAX_ATTEMPTS):
        pacer.wait()
        try:
            response = client.get(f"{BASE_URL}{path}", headers=HEADERS, params=params, timeout=30)
        except requests.RequestException:
            if attempt == MAX_ATTEMPTS - 1:
                raise
            time.sleep(BACKOFF_S[min(attempt, len(BACKOFF_S) - 1)])
            continue

        if response.ok:
            pacer.observe(response)
            return response.json()

        if response.status_code == 429:
            _sleep_to_minute_boundary()
            continue

        if response.status_code >= 500:
            if attempt == MAX_ATTEMPTS - 1:
                raise SyncError(response.status_code, response.json()["error"])
            time.sleep(BACKOFF_S[min(attempt, len(BACKOFF_S) - 1)])
            continue

        # 400, 401, 403, 404, 412 — retrying cannot change the answer.
        raise SyncError(response.status_code, response.json()["error"])

    raise RuntimeError(f"GET {path} gave up after {MAX_ATTEMPTS} attempts")


def parse_ts(value: str | None) -> datetime | None:
    """Timestamps are ISO 8601 with an offset, e.g. 2026-08-01T10:02:55.118307+00:00."""
    if not value:
        return None
    try:
        return datetime.fromisoformat(value)
    except ValueError:
        return None


def sync_transcript(client: requests.Session, conversation_id: str, pacer: Pacer) -> None:
    cursor: str | None = None
    while True:
        params: dict = {"limit": 200}
        if cursor:
            params["cursor"] = cursor

        page = get(client, f"/conversations/{conversation_id}/messages", params, pacer)
        # Upsert on message_id. A conversation you synced yesterday can gain
        # turns today, and you will read the old ones again.
        upsert_messages(conversation_id, page["items"])

        if not page["next_cursor"]:
            return                      # short page — the transcript is complete
        cursor = page["next_cursor"]


def sync() -> None:
    state = load_state()
    watermark = parse_ts(state.get("watermark"))
    floor = watermark - OVERLAP if watermark else None
    high = watermark

    pacer = Pacer()
    cursor: str | None = None
    done = False

    with requests.Session() as client:
        while not done:
            params: dict = {"limit": 100}
            if cursor:
                params["cursor"] = cursor

            page = get(client, "/conversations", params, pacer)

            for conversation in page["items"]:
                updated = parse_ts(conversation.get("updated_at"))

                # Ordered by updated_at DESC: the first old row ends the walk.
                if floor and updated and updated <= floor:
                    done = True
                    break

                upsert_conversation(conversation)
                sync_transcript(client, conversation["conversation_id"], pacer)

                if updated and (high is None or updated > high):
                    high = updated

            if done or not page["next_cursor"]:
                break                   # short page — no more conversations
            cursor = page["next_cursor"]

    # Only advance the watermark after the whole run succeeded. A crash
    # mid-walk leaves the old watermark in place and the next run redoes it.
    if high:
        save_state({"watermark": high.isoformat()})


# ---- your side ------------------------------------------------------------

def load_state() -> dict:
    try:
        with open("sync_state.json") as handle:
            return json.load(handle)
    except FileNotFoundError:
        return {}


def save_state(state: dict) -> None:
    with open("sync_state.json", "w") as handle:
        json.dump(state, handle)


def upsert_conversation(conversation: dict) -> None:
    ...   # keyed on conversation_id


def upsert_messages(conversation_id: str, messages: list[dict]) -> None:
    ...   # keyed on message_id


if __name__ == "__main__":
    sync()

Why the watermark is written last #

Notice that save_state runs once, after both loops finish. If the job dies on page four, the watermark is unchanged and the next run repeats pages one to four. Every write is an upsert, so the repeat is invisible in your data.

The alternative — advancing the watermark page by page — looks more efficient and loses data. A crash after the watermark is written but before the transcripts are stored leaves a permanent hole that no future run will revisit, because the conversations are now below the watermark.

Slow and repeatable beats fast and holed. This job is not the expensive part of your pipeline.

Why the transcript is re-read in full #

updated_at moved because something was said, but the API does not tell you what. There is no "messages since" filter — the messages cursor is a position, not a timestamp. So the job re-walks the transcript and upserts everything.

For a 12-turn conversation that is one request. For a 900-turn conversation it is five, at limit=200. If you have conversations long enough for that to matter, remember the last message_id you stored per conversation and stop the transcript walk once you reach it — the ids are time-ordered, so the check is a comparison rather than a lookup.

The first run is different #

The first run has no watermark, so it walks until a short page ends it. On a workspace with 20,000 conversations that is 200 list requests and up to 20,000 transcript requests — well over an hour at a safe pace, and enough to occupy your entire rate limit for the duration.

Treat the backfill as its own operation:

• Run it with its own API key so it cannot interfere with production traffic.
• Run it once, deliberately, not as the first tick of your five-minute schedule. A backfill that gets killed by a scheduler timeout and restarted from scratch every five minutes will never finish.
• Consider seeding the watermark by hand if you only care about recent history. Write a timestamp into your state store before the first run and the job will treat everything older as already synced.
• Expect 429s and let the pacer absorb them. They are not errors in this job; they are the shape of a backfill.

Webhooks and reconciliation: use both #

Neither mechanism is sufficient alone, and the reasons are different enough that combining them is not belt-and-braces — it is the design.

Polling alone is late. Your schedule is your latency. A five-minute sync means a conversation that ended thirty seconds ago is not in your store yet, which is fine for analytics and useless for anything that needs to react — routing, alerting, a live dashboard.

Webhooks alone are not your reconciliation database. Delivery is at-least-once. Retryable failures get four durable attempts and the final failed delivery can be replayed from the console. After 15 consecutive failed events the endpoint is disabled automatically; events created while it is disabled still need this scheduled sync after an admin re-enables it.

The production shape:

Two implementation notes where they meet:

• A webhook is a trigger, not a record. conversation.started fires when the conversation begins, so its transcript is empty or nearly so. Treat it as "this exists, and it is worth reading soon", not as data to store as-is.
• Deduplicate on the envelope id. The evt_… id at the top level of the webhook body is stable across redeliveries of the same event. This is separate from data.event_id, which is an internal identifier. Key your dedupe table on the top-level id or the Amomic-Event-Id header.

Getting events covers subscribing, and Retries and failures covers the ladder and the auto-disable in full.

Failures you will actually hit #

The branch that matters most is the one the sample code makes: retry 429, network errors, and 5xx; do not retry any other 4xx. A job that retries a 403 will hammer the API until someone notices, and it will not have synced anything.

Log error.request_id on every failure; quoting it to support turns a thread into a lookup. Every call your key made is also in the request log under Developers → Logs for 7 days, with its time, path, status and error code.

What is not in the synced data #

Worth knowing before someone asks your warehouse a question it cannot answer:

• Three roles. user is the customer, model is the agent (and, on a coexistence number, replies typed in the WhatsApp Business app), and human is a reply your team sent from the inbox or a ticket. The internal operational trail your staff see in the inbox is filtered out, and message_count excludes it too, so your row counts match message_count, not what a colleague counts in the console.
• No message metadata, no knowledge sources. You get message_id, role, text, created_at. Nothing about which document produced an answer.
• customer is as good as the channel it came from. WhatsApp gives you a platform identity, web chat gives you whatever the lead form collected, voice gives you the caller's number. An empty customer object is an ordinary case, not an error.
• No internal routing fields. Correlate through customer, or capture the correlation at the moment the conversation starts via the conversation.started webhook.
• Two timestamp shapes exist. REST resources use ISO 8601 with an offset and microseconds (2026-08-01T10:02:55.118307+00:00). The webhook envelope's created_at uses second precision with a literal Z (2026-08-01T10:02:55Z). If you store both, normalise on the way in.

Operating it #

• Alert on consecutive failed runs, not on a single one. One 429 storm or one 502 is noise. Three runs in a row that stored nothing is an incident.
• Emit the watermark as a metric. Watermark age is the one number that tells you the pipeline is healthy: if it stops advancing, the job is failing regardless of whether it is throwing.
• Watch the request log after a deploy. Developers → Logs shows every call this key made with its status, error code, and duration — the fastest way to see that your new pacing is doing what you think.
• Send X-Correlation-Id. Any value you like, echoed back on the response and recorded on our side, so a support thread can join your trace to ours.

Next #