The cloro API for Python

Quickstart

Install a HTTP client, store your API key in the environment, and make your first call. cloro absorbs 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 an authenticated request and a JSON parse.

pip install requests  # or: pip install httpx

Store your key in the environment — never hardcode secrets:

export CLORO_API_KEY="sk-..."

Authenticate and make the first call:

import os
import requests

API_KEY = os.environ["CLORO_API_KEY"]
BASE_URL = "https://api.cloro.dev/v1"

HEADERS = {
    "Authorization": f"Bearer {API_KEY}",
    "Content-Type": "application/json",
}

def search_google(query: str, country: str = "us") -> dict:
    """Single Google SERP call via cloro. Returns parsed JSON."""
    payload = {
        "query": query,
        "country": country,
        "include": ["organic", "ai_overview", "people_also_ask"],
    }
    resp = requests.post(
        f"{BASE_URL}/monitor/google",
        headers=HEADERS,
        json=payload,
        timeout=30,
    )
    resp.raise_for_status()
    return resp.json()

print(search_google("best running shoes"))

Every engine, one client

Google Search and every AI engine share the same auth header, request shape, and credit pool. You add coverage by changing the endpoint path, not by writing a new scraper per engine.

Base URL: https://api.cloro.dev. Full request and response schemas live in the API docs.

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
import os
from concurrent.futures import ThreadPoolExecutor, as_completed
from datetime import date

import requests

API_KEY = os.environ["CLORO_API_KEY"]
BASE_URL = "https://api.cloro.dev/v1"
HEADERS = {
    "Authorization": f"Bearer {API_KEY}",
    "Content-Type": "application/json",
}

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:
    payload = {
        "query": query,
        "country": country,
        "include": ["organic", "ai_overview"],
    }
    resp = requests.post(
        f"{BASE_URL}/monitor/google",
        headers=HEADERS,
        json=payload,
        timeout=30,
    )
    resp.raise_for_status()
    data = resp.json()
    positions = [
        item["position"]
        for item in data.get("organic", [])
        if TARGET_DOMAIN in item.get("url", "")
    ]
    return {
        "date": str(date.today()),
        "query": query,
        "country": country,
        "rank": positions[0] if positions else None,
        "in_ai_overview": TARGET_DOMAIN
        in str(data.get("ai_overview", {}).get("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.

import os
import requests

API_KEY = os.environ["CLORO_API_KEY"]
BASE_URL = "https://api.cloro.dev/v1"
HEADERS = {
    "Authorization": f"Bearer {API_KEY}",
    "Content-Type": "application/json",
}

# Same auth, same request shape — only the endpoint path changes per engine.
AI_ENGINES = {
    "chatgpt":    f"{BASE_URL}/monitor/chatgpt",
    "perplexity": f"{BASE_URL}/monitor/perplexity",
    "gemini":     f"{BASE_URL}/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, endpoint: str) -> dict:
    payload = {
        "prompt": prompt,
        "country": "us",
        "include": ["answer", "sources", "brand_mentions"],
    }
    resp = requests.post(endpoint, headers=HEADERS, json=payload, timeout=60)
    resp.raise_for_status()
    data = resp.json()
    sources = data.get("sources", [])
    return {
        "engine": engine,
        "prompt": prompt,
        "brand_cited": any(TARGET_BRAND in s for s in sources),
        "answer_snippet": data.get("answer", "")[:200],
    }


def run_ai_visibility_audit() -> list[dict]:
    results = []
    for prompt in BRAND_QUERIES:
        for engine, endpoint in AI_ENGINES.items():
            try:
                row = check_ai_visibility(prompt, engine, endpoint)
                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:

• Retry with backoff. Check for HTTP 429 and retry with exponential backoff plus jitter rather than hammering the API.
• 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.
• Go async for volume. When you need hundreds of queries per minute, httpx.AsyncClient with asyncio beats threads. Make sure you await every call.
• 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.

A minimal async client with backoff and a concurrency cap looks like this:

import asyncio
import os
import random

import httpx

API_KEY = os.environ["CLORO_API_KEY"]
BASE_URL = "https://api.cloro.dev/v1"
HEADERS = {"Authorization": f"Bearer {API_KEY}", "Content-Type": "application/json"}

# Cap concurrency at the plan limit — a Semaphore, not the number of tasks.
SEM = asyncio.Semaphore(10)


async def search(client: httpx.AsyncClient, query: str, max_retries: int = 3) -> dict:
    payload = {"query": query, "country": "us", "include": ["organic"]}
    delay = 1.0
    for attempt in range(max_retries):
        async with SEM:
            resp = await client.post(f"{BASE_URL}/monitor/google", json=payload)
        if resp.status_code == 429 and attempt < max_retries - 1:
            await asyncio.sleep(delay + random.uniform(0, delay * 0.5))
            delay *= 2  # exponential backoff with jitter
            continue
        resp.raise_for_status()
        return resp.json()
    raise RuntimeError("max retries exceeded")


async def main(queries: list[str]) -> list[dict]:
    async with httpx.AsyncClient(headers=HEADERS, timeout=30) as client:
        return await asyncio.gather(*(search(client, q) for q in queries))


if __name__ == "__main__":
    asyncio.run(main(["best running shoes", "trail shoes", "waterproof boots"]))

Error handling

The API returns conventional HTTP status codes. Branch on them once in a wrapper so every call site gets the same behaviour — retry the transient ones, fail fast on the rest.

import requests

def call(endpoint: str, payload: dict) -> dict:
    try:
        resp = requests.post(endpoint, headers=HEADERS, json=payload, timeout=30)
        resp.raise_for_status()
        return resp.json()
    except requests.HTTPError as exc:
        status = exc.response.status_code
        if status == 401:
            raise RuntimeError("Invalid or missing CLORO_API_KEY") from exc
        if status == 400:
            raise RuntimeError(f"Bad request: {exc.response.text}") from exc
        if status == 429:
            raise RuntimeError("Rate limit — back off and retry") from exc
        raise  # 5xx: transient, safe to retry with backoff
    except requests.Timeout as exc:
        raise RuntimeError("Request timed out — retry with backoff") from exc