2026-09-15 — Rank Tracking · Tutorials

Building a rank-tracking pipeline for AI answers with batches

A rank tracker for AI answers asks the same questions on a schedule and records where your domain appears among the cited sources. This post builds one on async batches: a submission that is safe to re-run, collection by webhook, and a reconciliation pass for anything that didn’t arrive.

The code is Python with httpx and reads API_URL (your API base URL) and API_KEY from the environment.

The unit of work

One task is one prompt, in one market, on one day. That triple is also its identity, so it makes the idempotency key:

def task_key(prompt_id: str, country: str, day: str) -> str:
    return f"{prompt_id}-{country}-{day}"

Keys are unique across your account, so re-running a day’s submission creates nothing twice: tasks that exist come back as RESOURCE_ALREADY_EXISTS.

Submitting in batches

POST /v1/async/task/batch takes 1 to 500 tasks. Each is validated and admitted on its own, and results answers each by index, in input order.

import os
import httpx

MAX_BATCH = 500
api = httpx.Client(
    base_url=os.environ["API_URL"],
    headers={"Authorization": f"Bearer {os.environ['API_KEY']}"},
    timeout=30,
)

def build_tasks(prompts: dict, countries: list, day: str, webhook_url: str) -> list:
    return [
        {
            "taskType": "CHATGPT",
            "payload": {"prompt": prompt, "country": country},
            "idempotencyKey": task_key(prompt_id, country, day),
            "webhook": {"url": webhook_url},
        }
        for prompt_id, prompt in prompts.items()
        for country in countries
    ]

def submit(tasks: list) -> dict:
    """Queues every task; returns the ids of tasks created by this call, by idempotency key."""
    created = {}
    for start in range(0, len(tasks), MAX_BATCH):
        chunk = tasks[start:start + MAX_BATCH]
        res = api.post("/v1/async/task/batch", json=chunk)
        res.raise_for_status()
        for item in res.json()["results"]:
            key = chunk[item["index"]]["idempotencyKey"]
            if item["success"]:
                created[key] = item["task"]["id"]
            elif item["error"]["code"] == "INSUFFICIENT_CREDITS":
                raise RuntimeError(f"out of credits at {key}")
            elif item["error"]["code"] != "RESOURCE_ALREADY_EXISTS":
                print("rejected", key, item["error"])
    return created

Persist what submit returns; reconciliation needs the ids. Three failures to plan for:

• INSUFFICIENT_CREDITS on an item. Tasks are checked in order against your balance less what open tasks have reserved, so later tasks will likely fail too. Stop, and re-run after credits reset or an upgrade; the keys skip what was already queued.
• 429 QUEUE_LIMIT_EXCEEDED for the whole request. The batch would overflow your queue, so none of it was queued. details.remainingCapacity says how many tasks fit now; wait for the queue to drain and re-run.
• VALIDATION_ERROR on an item. The input is wrong, for example an unsupported country. Fix it; resubmitting won’t help.

To let some prompts jump ahead, such as a launch week’s, add "priority": 10 to their tasks. The default is 1, and priorities only order your own queue.

Collecting results

Each finished task is POSTed to the webhook with the body GET /v1/async/task/{id} returns. The idempotency key rides along in task.idempotencyKey, so the receiver can file a result without a lookup:

from typing import Optional
from urllib.parse import urlsplit

def citation_rank(result: dict, domain: str) -> Optional[int]:
    """Position of the first cited source on the domain or a subdomain of it."""
    for source in sorted(result.get("sources", []), key=lambda s: s["position"]):
        host = urlsplit(source["url"]).hostname or ""
        if host == domain or host.endswith("." + domain):
            return int(source["position"])
    return None

def record(delivery: dict, domain: str, brand: str) -> dict:
    task = delivery["task"]
    row = {"key": task["idempotencyKey"], "task_id": task["id"], "status": task["status"]}
    if task["status"] == "COMPLETED":
        result = delivery["response"]["result"]
        row["rank"] = citation_rank(result, domain)
        row["mentioned"] = any(brand.lower() in e["name"].lower() for e in result.get("entities", []))
    return row

Verify the signature before calling record (webhooks guide), upsert rows on task_id because deliveries can repeat, and answer 2xx before slow work.

A FAILED task is not charged and its response holds the error. Keep its row so gaps in coverage stay visible. Its key stays taken, so a second attempt needs a new key, for example with an attempt suffix.

Reconciling

Webhook deliveries are retried with backoff for up to 10 attempts. If your endpoint is down longer, they stop, so after each run poll the tasks you created but never stored:

def reconcile(created: dict, stored_ids: set, domain: str, brand: str) -> list:
    rows = []
    for task_id in set(created.values()) - stored_ids:
        res = api.get(f"/v1/async/task/{task_id}")
        res.raise_for_status()
        delivery = res.json()
        if delivery["task"]["status"] in ("COMPLETED", "FAILED"):
            rows.append(record(delivery, domain, brand))
    return rows

Scheduling and cost

Start the submission from any scheduler once per period. A run costs prompts × markets × the engine’s price per task, failures are free, and the pricing page lists prices and add-ons. Check GET /v1/credits before a large run.

Keep reading