Python SDK

Install

pip install answerline

Example

import os
from answerline import Client

client = Client(os.environ["API_KEY"])
res = client.monitor.chatgpt(
    prompt="What are the best running shoes for flat feet?",
    country="US",
)
print(res["result"]["text"])

Mirrors the REST contract: client.monitor.chatgpt(...) is POST /v1/monitor/chatgpt, client.async_tasks.create(...) is POST /v1/async/task. Keyword arguments map one-to-one to request fields.

Runs anywhere Python 3.10+ runs: cron jobs, notebooks, Airflow, serverless functions. A synchronous Client and an asyncio AsyncClient share the same methods.

SDK and plain HTTP

The Python SDK is on PyPI: pip install answerline (Python 3.10+, built on httpx). Every SDK call maps to one REST request, so the same work runs over plain HTTP:

A minimal REST client:

import os
import httpx

api = httpx.Client(
    base_url="https://api.answerline.dev",
    headers={"Authorization": f"Bearer {os.environ['API_KEY']}"},
    timeout=httpx.Timeout(30.0, read=360.0),  # sync calls can take up to five minutes
)

res = api.post("/v1/monitor/chatgpt", json={"prompt": "Best CRM for a 5-person team?", "country": "US"})
res.raise_for_status()
result = res.json()["result"]
print(result["text"])
for source in result.get("sources", []):
    print(source["position"], source["url"])

The rest of this guide documents the SDK’s interface (pip install answerline). Code written against the REST shapes carries over unchanged.

Configuration

The SDK reads no environment variables. Client takes its settings as arguments:

import os
from answerline import Client

client = Client(
    os.environ["API_KEY"],                 # api_key, required
    base_url="https://api.answerline.dev", # default
    timeout=30.0,                          # seconds per attempt, most calls
    sync_timeout=360.0,                    # seconds per attempt, monitor calls and task creation
    max_retries=2,                         # retries after the first attempt
)

A client holds one httpx connection pool and can be shared across threads. Close it with close() or a with block. with_options(timeout=..., sync_timeout=..., max_retries=...) returns a copy that shares the pool with different settings; closing the copy does nothing, so close the original.

with Client(os.environ["API_KEY"]) as client:
    patient = client.with_options(max_retries=5)
    print(patient.credits())  # {"remaining": ..., "perCycle": ..., "cycleResetsAt": ...}

AsyncClient takes the same arguments, exposes the same methods as coroutines, and closes with aclose() or async with.

Synchronous monitor calls

Monitor calls return the parsed JSON dict, {"success": True, "result": {...}}. Keyword arguments map one to one to request fields, so anything in an engine’s request schema can be passed through **options.

answer = client.monitor.chatgpt(
    prompt="Which accounting software do freelancers recommend?",
    country="US",
    include={"markdown": True, "searchQueries": True},
)
result = answer["result"]
print(result["markdown"])
print(result.get("searchQueries"))

The client.monitor namespace has chatgpt, perplexity, gemini, copilot and grok (each prompt, country, **options), aimode(prompt, country=None, gl=None, **options), google(query=None, country=None, **options) and google_news(query, country=None, gl=None, **options). Calling the namespace directly is the generic form: client.monitor("gemini", prompt=..., country=...). Top-level shortcuts also exist: client.chatgpt, client.perplexity, client.gemini, client.copilot, client.grok, client.ai_mode, client.google and client.google_news. Perplexity and Grok are paused for now and coming back soon; until then calls to them are refused and not billed.

aimode and google_news raise ValueError before sending when neither country nor gl is given.

Google News, Google Search and AI Mode

news = client.monitor.google_news("electric vehicle tariffs", country="US", hl="en", pages=2)
for row in news["result"]["newsResults"]:
    print(row["position"], row["source"], row["title"])

Google Search and AI Mode follow the same pattern: client.monitor.google(query=..., country="US", pages=2, include={"aioverview": {"markdown": True}}) returns organicResults and the AI Overview, and client.monitor.aimode(prompt, gl="us", hl="en") returns the AI Mode answer.

client.google_goto(url) resolves a Google redirect link to its destination URL with a 15-second timeout. client.countries("chatgpt") lists supported country codes, and client.states("US") lists US state codes for state targeting on Gemini.

Credits per call

Synchronous calls add 7 credits to the engine’s price, and on Google Search and Google News 18.9 more per results page; tasks do not. ChatGPT is 6.2 credits as a task and 13.2 synchronously, plus 7 once if you enable rawResponse or shopping; searchQueries and ads are free. Gemini is 5, Copilot 47.3 and AI Mode 37.8. Google Search is 18.9 for the first page, 18.9 per extra page and 18.9 for the AI Overview. Google News is 2 for the first page and 2 per extra page. The credits guide lists every engine.

Async tasks

created = client.create_task(
    "CHATGPT",
    {"prompt": "Best CRM for a 5-person team?", "country": "US"},
    priority=5,
    idempotency_key="crm-us-2026-09-17",
    webhook_url="https://example.com/hooks/answers",
)
task_id = created["task"]["id"]

create_task(task_type, payload, priority=None, idempotency_key=None, webhook_url=None) answers immediately. Without an idempotency_key the SDK generates one, so a retried creation cannot create a duplicate. For recurring jobs, build the key from what the task means (prompt, market, day) so a rerun of the whole job is also a no-op.

Polling to completion

done = client.wait_for_task(task_id, timeout=660, interval=10.0)
if done["task"]["status"] == "COMPLETED":
    print(done["credits"]["creditsCharged"], done["response"]["result"]["text"])
else:
    print("failed, not charged:", done.get("response"))

wait_for_task polls get_task with jittered backoff from 0.5 s up to interval, polls through rate limits, 5xx and connection errors, and raises the built-in TimeoutError after timeout seconds (default 660). The task keeps running after a timeout.

The client.async_tasks namespace wraps the same operations: create, create_batch, get, wait, status, clear_queue, and run, which creates and waits in one call:

done = client.async_tasks.run("GOOGLE_NEWS", {"query": "crm software", "country": "US"}, timeout=600)

Batches

create_batch sends 1 to 500 tasks (MAX_BATCH) in one request and raises ValueError outside that range. Items use the wire’s camelCase keys; async_tasks.create_batch also accepts task_type, idempotency_key and webhook_url.

from datetime import date
from answerline import MAX_BATCH

day = date.today().isoformat()
prompts = ["best running shoes for flat feet", "best trail running shoes"]
tasks = [
    {
        "task_type": engine,
        "payload": {"prompt": prompt, "country": "US"},
        "idempotency_key": f"shoes-{i}-{engine}-US-{day}",
        "webhook_url": "https://example.com/hooks/answers",
    }
    for i, prompt in enumerate(prompts)
    for engine in ("CHATGPT", "GEMINI")
]

for start in range(0, len(tasks), MAX_BATCH):
    batch = client.async_tasks.create_batch(tasks[start:start + MAX_BATCH])
    for item in batch["results"]:
        if not item["success"] and item["error"]["code"] != "RESOURCE_ALREADY_EXISTS":
            print(item["index"], item["error"]["code"], item["error"]["message"])

Each item succeeds or fails independently with VALIDATION_ERROR, RESOURCE_ALREADY_EXISTS or INSUFFICIENT_CREDITS. Only a full queue rejects the whole request (429 QUEUE_LIMIT_EXCEEDED). client.async_status() reports queuedTasks, processingTasks, priorityBreakdown and concurrency; client.clear_queue() deletes tasks still QUEUED, which were never charged, and returns {"success": ..., "cleared": n}.

Webhooks

A task with a webhook URL is POSTed to that URL when it finishes. The body is the get_task JSON, and the Webhook-Signature header carries t=<unix seconds>,v1=<hex HMAC-SHA256>. Verify the raw bytes:

from answerline import verify_webhook, SIGNATURE_HEADER

ok = verify_webhook(raw_body, headers.get(SIGNATURE_HEADER), os.environ["WEBHOOK_SECRET"])

verify_webhook(body, header, secret, tolerance_seconds=300) compares in constant time and accepts either signature during a secret rotation. Without the SDK, the check is a few lines of standard library code; the AI visibility tracking guide has a Flask receiver that includes it.

Deduplicate deliveries on task.id, and ignore deliveries with "test": true, which the dashboard sends. The webhooks guide covers retries, rotation and redelivery.

Errors

from answerline import ApiError

try:
    client.monitor.chatgpt(prompt="Best CRM for a 5-person team?", country="US")
except ApiError as err:
    if err.code == "INSUFFICIENT_CREDITS":
        ...  # 403: top up or wait for the cycle to reset
    elif err.code == "CONCURRENT_LIMIT_EXCEEDED":
        ...  # every slot busy after retries: queue a task instead
    else:
        print(err.status, err.code, err.meta.request_id, err.body)
except (TimeoutError, ConnectionError) as err:
    print("transport failure:", err, "caused by", err.__cause__)

ApiError has status, code (the stable error code or None), body (parsed JSON or text) and meta, a frozen dataclass with status, headers, request_id, rate_limit_limit, rate_limit_remaining, concurrent_limit, concurrent_current, concurrent_remaining, credits_remaining, credits_charged and latency_ms. Transport errors are re-raised as built-in TimeoutError or ConnectionError, chained to the httpx exception. The full code list is in authentication and errors.

To read headers on a successful call or send your own request id, use request:

data, meta = client.request("GET", "/v1/credits", headers={"X-Request-Id": "nightly-run-42"})
print(data["remaining"], meta.rate_limit_remaining, meta.request_id)

Retries and timeouts

With max_retries=2 by default and capped exponential backoff with full jitter:

• 429 is always retried, honoring Retry-After up to 8 seconds.
• 5xx, timeouts and dropped connections are retried for GET and for task creation (which always has an idempotency key).
• Synchronous monitor calls are retried after a transport error only when no connection was made, because they may already be running and are charged on success. A 500 or 502 from them is final.
• Other 4xx are never retried.

Timeouts are httpx timeouts per attempt, applied to connecting, each read and write, and waiting for a pooled connection.

Recipes

FastAPI webhook receiver

import os
from fastapi import FastAPI, HTTPException, Request, Response
from answerline import verify_webhook

app = FastAPI()

@app.post("/hooks/answers")
async def answers(request: Request) -> Response:
    body = await request.body()
    if not verify_webhook(body, request.headers.get("webhook-signature"), os.environ["WEBHOOK_SECRET"]):
        raise HTTPException(status_code=400)
    delivery = await request.json()
    if not delivery.get("test"):
        ...  # upsert by delivery["task"]["id"], then enqueue slow work
    return Response(status_code=204)

Concurrent calls with AsyncClient

import asyncio, os
from answerline import AsyncClient

async def main() -> None:
    async with AsyncClient(os.environ["API_KEY"]) as client:
        prompt = "Which password manager is best for families?"
        answers = await asyncio.gather(
            client.monitor.chatgpt(prompt=prompt, country="US"),
            client.monitor.gemini(prompt=prompt, country="US"),
            return_exceptions=True,
        )
        for answer in answers:
            print(answer if isinstance(answer, Exception) else answer["result"]["text"][:200])

asyncio.run(main())

Each synchronous call holds a concurrency slot, so keep gather fan-out within your plan’s limit (rate limits); for larger fan-out, use a batch.

Airflow DAG

Submit a batch in one task and let webhooks deliver results, so the worker slot is not held for minutes:

import os
from datetime import datetime
from airflow.decorators import dag, task

@dag(schedule="@daily", start_date=datetime(2026, 9, 1), catchup=False)
def ai_answer_tracking():
    @task
    def submit(ds=None):
        from answerline import Client
        prompts = ["best CRM for small agencies", "HubSpot vs Pipedrive"]
        tasks = [
            {"taskType": engine, "payload": {"prompt": p, "country": "US"},
             "idempotencyKey": f"{engine}-{i}-US-{ds}",
             "webhook": {"url": os.environ["WEBHOOK_URL"]}}
            for engine in ("CHATGPT", "GEMINI")
            for i, p in enumerate(prompts)
        ]
        with Client(os.environ["API_KEY"]) as client:
            return client.create_batch(tasks)["summary"]

    submit()

ai_answer_tracking()

Keys use Airflow’s logical date ds, so a retried or cleared task instance recreates nothing. The same structure works as a cron script or a notebook cell. For batch pipeline design, see rank tracking pipelines with batches.

The quickstart shows the same first calls in curl and Python side by side.

Other integrations