Get Gemini citations in Python

The whole task is one HTTP call: requests.post to https://api.agentgeo.org/v1/fetches with {"query": ..., "surfaces": ["gemini"]}, a timeout=200, and raise_for_status() — then read answers[0]["sources"]. Everything after that is what you do with the list: normalize the URLs, count the hosts, and write the rows somewhere with their fetchedAt so next month has something to compare against. Below is the minimal call, then the script that does all three.

The minimal call

Two prerequisites: Python 3.9+ and requests. There is no SDK to install and no client object to configure — the API is a single endpoint, and the payload for Gemini is exactly two fields.

import requests

resp = requests.post(
    "https://api.agentgeo.org/v1/fetches",
    json={
        "query": "best postgres backup tool for small teams",
        "surfaces": ["gemini"],
    },
    headers={"Authorization": "Bearer ag_live_your_key_here"},
    timeout=200,  # the API holds slow surfaces up to 180s - stay above it
)
resp.raise_for_status()

answer = resp.json()["answers"][0]
print(answer["answerText"][:300])
for src in answer.get("sources") or []:
    print(src["position"], src["url"])

A grounded run comes back like this, trimmed:

{
  "id": "run_3f9a71c04b2e",
  "query": "best postgres backup tool for small teams",
  "surfaces": ["gemini"],
  "status": "completed",
  "recordsDelivered": 1,
  "creditsCharged": 1,
  "answers": [
    {
      "surfaceKey": "gemini",
      "answerText": "For a small team the usual shortlist is pgBackRest, Barman and ...",
      "sources": [
        { "title": "pgBackRest - Reliable PostgreSQL Backup & Restore", "url": "https://pgbackrest.org/", "position": 1 },
        { "title": "Barman Documentation", "url": "https://docs.pgbarman.org/", "position": 2 }
      ],
      "fetchedAt": "2026-07-25T08:41:19Z"
    }
  ]
}

Two things in that record shape the rest of the page. First, sources can legitimately come back empty — Gemini only cites when it grounds, the record is still delivered and still costs its one credit, and retrying won't make citations appear. Second, the payload really is two fields. The endpoint accepts country, language and web_search, but none of them is honoured on the gemini surface — web_search is ChatGPT-only, language maps to AI Overviews' hl, and country steers AI Overviews and Copilot — so passing them here is dead weight that misleads whoever reads the code next. The same two-field call ports unchanged to Node.js and cURL.

From one call to a dataset

One call tells you who Gemini cited this morning. The questions that justify a script — which hosts keep recurring across a topic, what share of queries ground at all, what changed since the last run — need a query set, honest URL normalization and rows that survive the run. This version does all of it with the standard library plus requests:

import csv
import pathlib
from collections import Counter
from urllib.parse import urlsplit

import requests

API = "https://api.agentgeo.org/v1/fetches"
HEADERS = {"Authorization": "Bearer ag_live_your_key_here"}
OUT = pathlib.Path("gemini_citations.csv")

QUERIES = [
    "best postgres backup tool for small teams",
    "pgbackrest vs barman",
    "how to test postgres restores automatically",
    "postgres point in time recovery setup",
]


def normalize(url):
    """Split a cited URL into a (host, path) worth counting.

    Lowercase the host, drop a leading www., discard the query string -
    otherwise utm tags and trailing slashes split one page into four rows.
    """
    parts = urlsplit(url)
    host = parts.netloc.lower().removeprefix("www.")
    path = parts.path.rstrip("/") or "/"
    return host, path


hosts = Counter()
rows = []
delivered = grounded = 0

for query in QUERIES:
    resp = requests.post(
        API,
        json={"query": query, "surfaces": ["gemini"]},
        headers=HEADERS,
        timeout=200,  # the API holds slow surfaces up to 180s - stay above it
    )
    resp.raise_for_status()

    for answer in resp.json()["answers"]:
        if answer.get("status") == "failed":
            snap = (answer.get("providerFields") or {}).get("snapshot_id")
            print(f"failed at zero credits: {query!r} snapshot_id={snap}")
            continue

        delivered += 1
        sources = answer.get("sources") or []
        fetched_at = answer.get("fetchedAt")

        if not sources:
            # Ungrounded run. Keep the row - it is the grounding history.
            rows.append([fetched_at, query, "", 0, "", "", ""])
            continue

        grounded += 1
        for src in sources:
            host, path = normalize(src["url"])
            hosts[host] += 1
            rows.append([
                fetched_at,
                query,
                src["position"],
                len(sources),
                host,
                path,
                src["url"],
            ])

if delivered:
    print(f"grounding rate: {grounded}/{delivered} ({grounded / delivered:.0%})")
for host, n in hosts.most_common(10):
    print(f"{n:>3}  {host}")

# Append, never overwrite: the accumulated file is what next month diffs.
is_new = not OUT.exists()
with OUT.open("a", newline="") as f:
    writer = csv.writer(f)
    if is_new:
        writer.writerow(
            ["fetchedAt", "query", "position", "n_sources", "host", "path", "url"]
        )
    writer.writerows(rows)
print(f"appended {len(rows)} rows to {OUT}")

Three decisions in there carry the weight. normalize() collapses www.example.com/guide/?utm_source=x and example.com/guide into one countable page — skip it and the top-host table splits real winners into fragments. Ungrounded answers still get a row, because the share of empty rows is the grounding history, and dropping them silently turns every future rate calculation into 100%. And the file opens in append mode: overwrite it and you always hold exactly one snapshot, which is the one thing a snapshot can't be compared with. The n_sources column is there because position is only meaningful relative to its own answer — first of two and first of nine are different outcomes.

With a few runs accumulated, the questions worth asking are all cheap column operations:

Grounding is observed, not requested

On chatgpt, web_search is a request parameter: you decide whether the engine browses. On gemini the same field is ignored — the engine grounds when it judges the query needs current evidence and answers from its own weights when it doesn't. There is no knob. That inversion is why the grounding rate, not the host table, is the number this surface adds: a query whose runs start arriving with sources is a query Google has decided deserves fresh pages, which is exactly when citation work on the topic starts to pay. The task-level guide covers what to do with that signal; the script's job is to make sure the rate exists in your data at all.

The credit math makes a scheduled run easy to justify: a delivered record costs one credit whether it cited nine pages or none, and a failed record costs nothing. Twenty queries weekly is twenty credits weekly.

Where this goes wrong

• An empty sources[] on a delivered record is a result. Branch on the record's status first and len(sources) second, and write the empty runs to disk. Retrying until citations appear doesn't test the engine — it re-buys the same fact one credit at a time.
• Params that do nothing here. web_search is honoured only by chatgpt, language only by google_ai_overview, and country by google_ai_overview and copilot. None of the three touches gemini. Keep the payload at query plus surfaces, or the next person to read the script will assume the knob works.
• One page, four spellings. Tracking parameters, trailing slashes, www. and uppercase hosts split a single page across rows and deflate its count. Normalize at write time, not at report time — a CSV of raw URLs makes every future analysis redo the cleanup.
• A client timeout under the server's. The API holds a slow scrape for up to 180 seconds. Set timeout=200 so requests outlasts the server — give up at 60 and you abandon answers the API would have delivered at second 120.
• Scrapes that outrun the budget. A fetch that exceeds the 180-second window comes back status: "failed" at zero credits, with providerFields.snapshot_id on the record. A follow-up POST with top-level "snapshot_id" and the same single surface redeems the finished answer instead of paying for a re-scrape.
• A CSV opened with "w". Write mode keeps exactly today and deletes the history the whole exercise exists to build. Append, with fetchedAt on every row — and if you prefer one file per run, put the date in the filename, because file mtimes don't survive a copy.

Summary

One requests.post with a query and "surfaces": ["gemini"] returns the answer and its citations as structured JSON; the rest of the job is discipline. Normalize URLs before counting, keep the ungrounded runs, append instead of overwriting, and let the CSV accumulate until the grounding rate and the host table both have a trend. The identical script runs against the other five engines by changing one string — the Python API page covers the endpoint beyond this one task, and the docs list which optional field each surface honours.