The cloro API for Python

Quickstart

Install the SDK, store your API key in the environment, and make your first call. The cloro SDK wraps proxies, rendering, and anti-bot logic on the server — whether you're hitting the Google SERP API or an AI engine like ChatGPT — so your Python code shrinks to one method call and a parse. Working in Node or TypeScript? See the cloro API for JavaScript & Node.

pip install cloro

Store your key in the environment — never hardcode secrets:

export CLORO_API_KEY="sk-..."

Authenticate and make the first call:

from cloro import Cloro

# Reads your key from the CLORO_API_KEY environment variable.
client = Cloro()

# Google Search — structured SERP JSON, no proxies or selectors.
serp = client.monitor.google(query="best running shoes", country="US")
for item in serp["result"]["organicResults"]:
    print(item["position"], item["link"])

# Any AI engine is the same client — just change the method.
answer = client.monitor.chatgpt(
    prompt="What do you know about Acme Corp?",
    country="US",
)
print(answer["result"]["text"])

Every engine, one client

Google Search and every AI engine share one client and one credit pool. You add coverage by changing the method — each maps to one endpoint, shown below — not by writing a new scraper per engine.

Base URL: https://api.cloro.dev. Full request and response schemas live in the API docs, and the Python SDK reference documents every client method.

Recipe 1: rank tracker across countries

The most common production job: a nightly tracker that checks a keyword list across several markets and writes results to JSONL you can load into pandas, BigQuery, or a database. It fans out 300 queries (100 keywords × 3 countries) across a capped pool of workers. For the endpoint-level detail, see the Google rank tracking API guide.

import json
from concurrent.futures import ThreadPoolExecutor, as_completed
from datetime import date

from cloro import Cloro

# The client reads CLORO_API_KEY and retries 429/5xx with backoff for you.
client = Cloro()

KEYWORDS = [
    "best running shoes",
    "trail running shoes men",
    "waterproof running shoes",
    # ... up to 100
]
COUNTRIES = ["US", "GB", "DE"]
TARGET_DOMAIN = "nike.com"
MAX_WORKERS = 10  # match your plan's concurrency limit, not your CPU count


def fetch_one(query: str, country: str) -> dict:
    res = client.monitor.google(
        query=query,
        country=country,
        include={"aioverview": {"markdown": True}},
    )
    result = res["result"]
    positions = [
        item["position"]
        for item in result["organicResults"]
        if TARGET_DOMAIN in item.get("link", "")
    ]
    aio_sources = str(result.get("aioverview", {}).get("sources", []))
    return {
        "date": str(date.today()),
        "query": query,
        "country": country,
        "rank": positions[0] if positions else None,
        "in_ai_overview": TARGET_DOMAIN in aio_sources,
    }


def run_tracker(output_path: str = "ranks.jsonl") -> None:
    tasks = [(kw, c) for kw in KEYWORDS for c in COUNTRIES]
    rows: list[dict] = []

    with ThreadPoolExecutor(max_workers=MAX_WORKERS) as pool:
        futures = {pool.submit(fetch_one, kw, c): (kw, c) for kw, c in tasks}
        for future in as_completed(futures):
            kw, c = futures[future]
            try:
                rows.append(future.result())
                print(f"OK  {kw} / {c}")
            except Exception as exc:
                print(f"ERR {kw} / {c}: {exc}")

    with open(output_path, "w") as f:
        for row in rows:
            f.write(json.dumps(row) + "\n")

    print(f"\nDone. {len(rows)} rows written to {output_path}")


if __name__ == "__main__":
    run_tracker()

SERP calls are I/O-bound, so threads are the right tool here — not multiprocessing. Cap MAX_WORKERS at your plan's concurrency limit, and keep each returned row flat so it writes cleanly to JSONL, CSV, or a column.

Recipe 2: AI visibility across ChatGPT, Perplexity & Gemini

Rank tracking no longer stops at Google. If your brand appears in a ChatGPT answer but not a Perplexity citation, that gap is worth measuring. Each AI engine is a separate endpoint on the same credit pool — notice how little changes from the rank-tracker code above.

from cloro import Cloro

client = Cloro()

# Same client, same call shape — only the method changes per engine.
ENGINES = {
    "chatgpt": client.monitor.chatgpt,
    "perplexity": client.monitor.perplexity,
    "gemini": client.monitor.gemini,
}

BRAND_QUERIES = [
    "best electric SUV 2026",
    "tesla model y vs competitors",
    "ev range comparison",
]
TARGET_BRAND = "tesla.com"


def check_ai_visibility(prompt: str, engine: str, call) -> dict:
    result = call(prompt=prompt, country="US")["result"]
    sources = result.get("sources", [])
    return {
        "engine": engine,
        "prompt": prompt,
        "brand_cited": any(TARGET_BRAND in (s.get("url") or "") for s in sources),
        "answer_snippet": result.get("text", "")[:200],
    }


def run_ai_visibility_audit() -> list[dict]:
    results = []
    for prompt in BRAND_QUERIES:
        for engine, call in ENGINES.items():
            try:
                row = check_ai_visibility(prompt, engine, call)
                results.append(row)
                status = "cited" if row["brand_cited"] else "not cited"
                print(f"[{engine}] {prompt[:40]}... -> {status}")
            except Exception as exc:
                print(f"ERR [{engine}] {prompt[:40]}...: {exc}")
    return results


if __name__ == "__main__":
    rows = run_ai_visibility_audit()
    cited = sum(1 for r in rows if r["brand_cited"])
    print(f"\n{cited}/{len(rows)} queries cite {TARGET_BRAND} across AI engines")

This is the foundation of AI visibility monitoring: tracking where your brand surfaces (or doesn't) in generative answers alongside traditional search. You add engines by adding endpoints, not by writing new scrapers.

Production concerns

Five things separate a script that works once from one that runs on a schedule:

• Retries are built in. The client retries 429 and 5xx responses with exponential backoff automatically — tune it with Cloro(max_retries=...).
• Cap concurrency. Match your worker count to your plan's concurrency limit — a ThreadPoolExecutor size for sync code, an asyncio.Semaphore for async. Set it to the plan limit, not your CPU count.
• Scale with the task queue. When you need hundreds of queries per minute, enqueue them with client.async_tasks.create_batch() and poll — the queue absorbs concurrency and retries instead of you hand-tuning a Semaphore.
• Keep secrets out of code. Load the key from an environment variable or a .env file with python-dotenv, add .env to .gitignore, and rotate any key that lands in a commit.
• Access fields defensively. Use .get() and handle missing keys — response shapes evolve, and not every query returns an AI Overview box.

Batching high-volume work through the async task queue looks like this:

from cloro import Cloro

client = Cloro()

KEYWORDS = ["best running shoes", "trail shoes", "waterproof boots"]

# Enqueue up to 500 tasks in one request, then poll each to completion.
# The queue absorbs concurrency and retries — no Semaphore to hand-tune.
results = client.async_tasks.create_batch(
    [{"task_type": "GOOGLE", "payload": {"query": kw, "country": "US"}} for kw in KEYWORDS]
)

for item in results:
    if item["success"]:
        done = client.async_tasks.wait(item["task"]["id"])
        top = done["response"]["result"]["organicResults"][0]
        print(item["task"]["id"], "->", top["link"])
    else:
        print("failed:", item["error"]["message"])

Error handling

The SDK raises typed exceptions, all subclassing CloroError. Catch the ones you want to handle specially and let the transient ones (which the client already retries) bubble. They map to the underlying HTTP status codes:

from cloro import (
    Cloro,
    AuthenticationError,
    BadRequestError,
    RateLimitError,
    CloroError,
)

client = Cloro()

try:
    res = client.monitor.google(query="best running shoes", country="US")
except AuthenticationError as exc:
    raise RuntimeError("Invalid or missing CLORO_API_KEY") from exc
except BadRequestError as exc:
    raise RuntimeError(f"Bad request: {exc}") from exc
except RateLimitError:
    ...  # the client already retried; back off further if this persists
except CloroError:
    raise  # timeouts, 5xx, task failures — all subclass CloroError