Bulk SERP API: batch Google searches, synchronously or as queued jobs

curl -X POST "https://api.searlo.tech/api/v1/serp/batch" \
  -H "x-api-key: YOUR_API_KEY" \
  -H "Content-Type: application/json" \
  -d '{
    "keywords": ["best running shoes", "running shoes for flat feet", "trail running shoes"],
    "gl": "us",
    "hl": "en",
    "device": "mobile"
  }'

const res = await fetch("https://api.searlo.tech/api/v1/serp/batch", {
  method: "POST",
  headers: { "x-api-key": "YOUR_API_KEY", "Content-Type": "application/json" },
  body: JSON.stringify({
    keywords: ["best running shoes", "running shoes for flat feet", "trail running shoes"],
    gl: "us",
    hl: "en",
  }),
});
const batch = await res.json();

for (const row of batch.results) {
  if (!row.success) {
    console.warn(`${row.keyword}: ${row.message} (retryable: ${row.retryable})`);
    continue;
  }
  const top = row.organic[0];
  console.log(row.keyword, "->", top?.domain, `(${row.peopleAlsoAsk.length} PAA questions)`);
}
console.log(`credits used: ${batch.credits.used}, failed rows: ${batch.meta.failed}`);

import requests

resp = requests.post(
    "https://api.searlo.tech/api/v1/serp/batch",
    json={
        "keywords": ["best running shoes", "running shoes for flat feet"],
        "gl": "us",
        "hl": "en",
        "aioverview": True,  # off by default in a batch; no extra credits
    },
    headers={"x-api-key": "YOUR_API_KEY"},
    timeout=600,  # the response arrives when every keyword is done
)
resp.raise_for_status()
batch = resp.json()

for row in batch["results"]:
    has_aio = bool(row.get("aioverview"))
    print(row["keyword"], row["success"], len(row.get("organic", [])), "AI Overview:", has_aio)

{
  "schema": "serp-batch.v1",
  "results": [
    {
      "keyword": "best running shoes",
      "success": true,
      "searchParameters": { "q": "best running shoes", "gl": "us", "hl": "en", "page": 1, "pages": 1, "num": 10, "device": "desktop", "type": "serp" },
      "organic": [
        { "position": 1, "title": "The best running shoes of 2026, tested", "url": "https://www.example-runner.com/best-running-shoes", "domain": "example-runner.com", "snippet": "…", "page": 1 }
      ],
      "ads": [],
      "videos": [],
      "peopleAlsoAsk": [{ "question": "Which running shoe brand is best?" }],
      "relatedSearches": [{ "query": "best running shoes for beginners" }],
      "knowledgeGraph": null,
      "aiModeAvailable": true,
      "fetchedAt": "2026-09-29T11:20:07.331Z",
      "cached": false
    },
    {
      "keyword": "trail running shoes",
      "success": false,
      "status": 429,
      "retryable": true,
      "message": "Google SERP is at capacity. Retry in 60s — immediate retries will be refused."
    }
  ],
  "credits": { "used": 9 },
  "meta": {
    "requested": 3,
    "distinct": 3,
    "succeeded": 2,
    "failed": 1,
    "pages": 1,
    "gl": "us",
    "hl": "en",
    "device": "desktop",
    "max_sync_batch": 20
  }
}

import hashlib
import json
import time

import requests

API = "https://api.searlo.tech/api/v1"
HEADERS = {"x-api-key": "YOUR_API_KEY"}
RUN = "2026-09-29"  # part of each idempotencyKey: re-running today re-uses the same tasks

with open("keywords.txt", encoding="utf-8") as f:
    keywords = list(dict.fromkeys(line.strip() for line in f if line.strip()))[:500]


def key_for(kw):
    return f"serp:us:en:{RUN}:" + hashlib.sha1(kw.encode()).hexdigest()  # keys max 128 chars


# 1. Queue everything in one request (up to 500 tasks per call).
tasks = [
    {
        "endpoint": "serp",
        "request": {"q": kw, "gl": "us", "hl": "en", "pages": 1, "aioverview": False},
        "idempotencyKey": key_for(kw),
    }
    for kw in keywords
]
queued = requests.post(f"{API}/async/task/batch", json={"tasks": tasks}, headers=HEADERS, timeout=120).json()
if "tasks" not in queued:
    raise SystemExit(f"batch rejected: {queued.get('message')}")

pending = {}  # taskId -> keyword, in queue order
for kw, item in zip(keywords, queued["tasks"]):
    if item["accepted"]:
        pending[item["taskId"]] = kw
    else:
        print("not queued:", kw, "-", item["error"])
for kw in keywords[len(queued["tasks"]):]:
    print("not queued (queueing stopped, usually out of credits):", kw)
print(f"queued {queued['accepted']} of {queued['total']}")

# 2. Poll until every task has finished.
with open("serps.jsonl", "a", encoding="utf-8") as out:
    while pending:
        time.sleep(20)
        for task_id, kw in list(pending.items()):
            task = requests.get(f"{API}/async/task/{task_id}", headers=HEADERS, timeout=30).json()["task"]
            if task["status"] == "COMPLETED":
                out.write(json.dumps({"keyword": kw, "serp": task["result"]}) + "\n")
                del pending[task_id]
            elif task["status"] in ("FAILED", "CANCELLED"):
                print("gave up:", kw, task.get("error"))  # its credits are refunded
                del pending[task_id]
            elif task["status"] == "QUEUED":
                break  # tasks run roughly in queue order: check the rest on the next pass
        print(f"{len(pending)} tasks still to finish")