2026-09-17, News · Brand Monitoring · Tutorials

Google Alerts API: there isn't one. Build programmatic alerts instead

There is no official Google Alerts API. Google documents Alerts only as an email product, with no developer endpoint, key or SDK (checked 2026-09-17). If you need alerts inside your own system, a Slack channel, a CRM or a screening workflow, you build them: run the queries an alert would run on a schedule, store what you have seen, and notify on anything new. Below are the options, then a working pipeline on this API’s Google News and Google Search endpoints with signed webhooks.

What Google offers

Google Alerts is a Google service that emails you when new results match a query. Google’s help page (support.google.com/websearch/answer/4815696, checked 2026-09-17) describes creating an alert and choosing how often, which sources, the language, the region, how many results and which account receives it. It says nothing about an API.

We looked for developer documentation on developers.google.com and found none for Alerts. The only “alerts” product there is Public Alerts, which is for publishing emergency alerts and unrelated.

Two consequences:

1. Anything sold as a “Google Alerts API” either automates the Alerts web interface on your behalf or collects results some other way. Neither is backed by Google.
2. RSS delivery, which many guides describe, is not documented on Google’s help page for Alerts. It appears in community forum threads and third-party articles. Do not build a business process on a delivery channel the vendor does not document.

The options for programmatic alerts

If the question your alert answers is “did Google surface something new about this entity”, the last option is the closest match, because it reads the same surfaces a person would check.

The pipeline

An alert pipeline has five parts:

1. Watchlist. Entities with the queries and markets to check.
2. Scheduler. Submits one task per query, market and window.
3. Receiver. Accepts signed webhook deliveries and stores them.
4. Deduper. Decides whether a result is new for that entity.
5. Notifier. Sends the new items somewhere people will see them.

Each part below is Python with requests and Flask. The API base is https://api.answerline.dev, and every request carries Authorization: Bearer <key>.

1. The watchlist

Keep queries explicit. Google Alerts hides its matching; you should not.

WATCHLIST = [
    {"id": "acme", "news": '"acme corp"', "search": '"acme corp" -site:acme.com', "markets": [("US", "en"), ("GB", "en")]},
    {"id": "acme-ceo", "news": '"jane doe" acme', "search": None, "markets": [("US", "en")]},
    {"id": "rival", "news": '"rival inc"', "search": None, "markets": [("US", "en"), ("DE", "de")]},
]

• Quote brand names so the results page matches the phrase.
• Exclude your own domain from web queries so every product page does not become an alert.
• Use the market’s language in hl; a German reader’s results for a brand are not a translation of the US page.

2. The scheduler

Each run submits one batch. POST /v1/async/task/batch accepts 1 to 500 tasks, validates each one independently, and returns a result per task by index.

• GOOGLE_NEWS tasks take query, a country as country or gl, and optionally hl and device.
• GOOGLE tasks take query and country in the standard shape, or a full Google search url. The URL shape keeps Google’s tbs time filter, which lets a web check ask for the past day (tbs=qdr:d), the setting behind “Past 24 hours” in Google’s Tools menu. Check that the results reflect the window for your queries before you rely on it.

import os
import urllib.parse
from datetime import datetime, timezone

import requests

API = "https://api.answerline.dev"
HEADERS = {"Authorization": f"Bearer {os.environ['ANSWERLINE_API_KEY']}"}
WEBHOOK = "https://alerts.example.com/hooks/results"

def window_id(now: datetime) -> str:
    return now.strftime("%Y-%m-%dT%H")  # one run per hour

def build_tasks(now: datetime) -> list[dict]:
    window = window_id(now)
    tasks = []
    for entity in WATCHLIST:
        for country, hl in entity["markets"]:
            tasks.append({
                "taskType": "GOOGLE_NEWS",
                "payload": {"query": entity["news"], "country": country, "hl": hl},
                "idempotencyKey": f"alert:news:{entity['id']}:{country}:{window}",
                "webhook": {"url": WEBHOOK},
            })
            if entity["search"] and now.hour % 6 == 0:  # web checks every six hours
                url = "https://www.google.com/search?" + urllib.parse.urlencode(
                    {"q": entity["search"], "gl": country.lower(), "hl": hl, "tbs": "qdr:d"}
                )
                tasks.append({
                    "taskType": "GOOGLE",
                    "payload": {"url": url},
                    "idempotencyKey": f"alert:web:{entity['id']}:{country}:{window}",
                    "webhook": {"url": WEBHOOK},
                })
    return tasks

def submit(tasks: list[dict]) -> None:
    for start in range(0, len(tasks), 500):
        chunk = tasks[start:start + 500]
        resp = requests.post(f"{API}/v1/async/task/batch", headers=HEADERS, json=chunk, timeout=60)
        resp.raise_for_status()
        for item in resp.json()["results"]:
            if not item["success"] and item["error"]["code"] != "RESOURCE_ALREADY_EXISTS":
                print("not queued:", chunk[item["index"]]["idempotencyKey"], item["error"]["code"])

if __name__ == "__main__":
    submit(build_tasks(datetime.now(timezone.utc)))

Run it from cron at the top of each hour. The idempotency key is built from what the task means (entity, surface, market, hour), so a cron job that fires twice or a retry after a timeout creates nothing twice: the repeat comes back as RESOURCE_ALREADY_EXISTS. A 429 QUEUE_LIMIT_EXCEEDED rejects the whole batch; wait and resubmit the same tasks. More on which errors to retry is in API errors and retries.

3. The receiver

When a task finishes, the API POSTs the same body GET /v1/async/task/{taskId} returns: task (with status COMPLETED or FAILED), credits, and response. Each delivery carries a Webhook-Signature header of the form t=<unix seconds>,v1=<hex>, where v1 is the HMAC-SHA256 of <t>.<raw body> keyed with your signing secret. During a secret rotation there can be two v1 values; accept either.

Verify the raw bytes before parsing, store, answer 2xx quickly, and process later. Deliveries that do not get a 2xx within 15 seconds are retried, up to 10 attempts.

import hashlib
import hmac
import json
import os
import sqlite3
import time

from flask import Flask, abort, request

app = Flask(__name__)
SECRET = os.environ["WEBHOOK_SECRET"].encode()
db = sqlite3.connect("alerts.db", check_same_thread=False)
db.executescript("""
create table if not exists inbox (task_id text primary key, body text not null, received_at real not null);
create table if not exists seen (entity text, market text, link text, first_seen real, title text, source text,
                                 primary key (entity, market, link));
""")

def verified(raw: bytes, header: str | None) -> bool:
    if not header:
        return False
    parts = [p.split("=", 1) for p in header.split(",") if "=" in p]
    stamps = [v for k, v in parts if k == "t"]
    sigs = [v for k, v in parts if k == "v1"]
    if len(stamps) != 1 or not sigs or abs(time.time() - int(stamps[0])) > 300:
        return False
    expected = hmac.new(SECRET, stamps[0].encode() + b"." + raw, hashlib.sha256).hexdigest()
    return any(hmac.compare_digest(expected, s) for s in sigs)

@app.post("/hooks/results")
def results():
    raw = request.get_data()
    if not verified(raw, request.headers.get("Webhook-Signature")):
        abort(400)
    body = json.loads(raw)
    if body.get("test"):
        return "", 204
    db.execute("insert or ignore into inbox values (?, ?, ?)", (body["task"]["id"], raw.decode(), time.time()))
    db.commit()
    return "", 204

Two details matter:

• Deduplicate deliveries on task.id. The same delivery can arrive more than once and in any order. insert or ignore on the primary key handles it.
• Ignore test deliveries. Test sends from the dashboard are validly signed and carry "test": true at the top level.

Webhook receiver design goes further on inbox tables, ordering and outages.

4. The deduper

The inbox holds whole results. The deduper turns them into “new for this entity in this market”. The task’s idempotency key already encodes entity and market, so parse it back out:

def extract(body: dict) -> tuple[str, str, list[dict]]:
    _, surface, entity, market, _window = body["task"]["idempotencyKey"].split(":", 4)
    if body["task"]["status"] != "COMPLETED":
        return entity, market, []
    result = body["response"]["result"]
    if surface == "news":
        rows = result.get("newsResults", [])
        return entity, market, [{"link": r["link"], "title": r["title"], "source": r.get("source", "")} for r in rows]
    rows = result.get("organicResults", [])
    return entity, market, [{"link": r["link"], "title": r["title"], "source": r.get("displayedLink", "")} for r in rows]

def new_items(body: dict) -> list[dict]:
    entity, market, rows = extract(body)
    fresh = []
    for row in rows:
        cur = db.execute(
            "insert or ignore into seen values (?, ?, ?, ?, ?, ?)",
            (entity, market, row["link"], time.time(), row["title"], row["source"]),
        )
        if cur.rowcount:
            fresh.append({**row, "entity": entity, "market": market})
    db.commit()
    return fresh

The maxsplit of 4 keeps the window intact even if you later add minutes to it.

Seed before you alert. The first run for a new entity returns a full page of results, all “new”. Run the watchlist once with notifications off so the seen table holds the baseline.

link is the resolved destination; when Google served a redirect, the redirect is in redirectLink and does not pollute your key.

5. The notifier

Send new items to the channel people already read. A Slack incoming webhook is a single POST:

def notify(items: list[dict]) -> None:
    if not items:
        return
    lines = [f"*{i['entity']}* ({i['market']}): <{i['link']}|{i['title']}> — {i['source']}" for i in items[:20]]
    requests.post(os.environ["SLACK_WEBHOOK_URL"], json={"text": "\n".join(lines)}, timeout=10)

def process_inbox() -> None:
    rows = db.execute("select task_id, body from inbox").fetchall()
    for task_id, raw in rows:
        notify(new_items(json.loads(raw)))
        db.execute("delete from inbox where task_id = ?", (task_id,))
        db.commit()

Run process_inbox every minute from a worker. For low-urgency entities, batch them into a daily digest instead of instant messages; alert fatigue kills more monitoring programs than missed articles do.

Reconciliation: the check Google Alerts never gave you

Email alerts fail silently. A pipeline can prove completeness. After each run’s expected finish time:

1. List the idempotency keys you submitted for the window.
2. Compare with the task ids in your inbox history.
3. For any missing, call GET /v1/async/task/{taskId} with the id you stored from the batch response, and ingest the body if it is COMPLETED or FAILED.
4. Resubmit FAILED tasks under a new key suffix if the window still matters. Failed tasks are not charged.

Store the task id from each batch response next to its key for this purpose. Tasks that have not started within 72 hours fail with QUEUE_WAIT_EXCEEDED, which an hourly alert should never approach, but the reconciliation will catch it if it happens.

What it costs

Credit costs: Google News async task 3 credits, Google Search async task 3 credits (plus 2 per extra results page, not used here). Synchronous calls add 2; the scheduler above uses async tasks only.

For the watchlist above (5 entity-market pairs on News, 2 on web):

• News, hourly: 5 × 24 × 3 = 360 credits a day.
• Web, every six hours: 2 × 4 × 3 = 24 credits a day.
• Total: 384 credits a day, about 11,500 a month.

The free tier’s 500 credits a month are enough to build and seed the pipeline and run it for a day or two; pricing lists the plans. Halving frequency halves the bill, so set cadence per entity by how fast its coverage moves.

Matching Google Alerts’ settings

Pitfalls

• Alerting on the baseline. Seed each new entity first.
• Keys with random parts. A key built from a UUID per attempt makes retries create duplicate tasks. Build it from entity, surface, market and window.
• Deduplicating deliveries on Webhook-Id. It is not covered by the signature; use task.id.
• Unquoted brand names. Common words match everything; quote phrases and add a disambiguating term.
• One market. Adverse coverage often breaks in the local press first. Add the markets where the entity operates, in their languages.
• No completeness check. Without reconciliation, a broken receiver looks like a quiet news day.

For entity screening at scale, see the adverse media screening use case; for the no-code version of the notifier, AI-answer alerts in Zapier shows the same pattern. Start with the Google News engine page or the webhooks docs.

Keep reading