I build Qovia, the API used in this post. The monitoring logic (search, de-duplicate, filter, alert) works with any Reddit data source; the state file, CSV and Slack parts are plain Python.
If you ship a developer tool or a SaaS product, Reddit is where people ask "is there an alternative to X?", complain about your onboarding, or recommend you to someone else, often without tagging you. Finding those threads early is worth a lot: a helpful answer in the first hour of a thread does more than a week of ads. Doing it by hand means refreshing Reddit search every morning, so let's automate it.
In this tutorial we'll write a script that:
Run it from cron every hour and you have a basic social-listening tool for a few dollars a month.
You need Python 3.9+ and requests:
pip install requests
export QOVIA_KEY=sk-qovia-... # key from the Qovia console
export SLACK_WEBHOOK_URL=https://hooks.slack.com/services/... # optional
Get a key from Qovia. Requests authenticate with Authorization: Bearer <key> (an x-api-key header also works). The Reddit search reference shows every parameter with a real response; the other Reddit endpoints are listed next to it.
You don't need a Reddit account, OAuth app or client secret for this. The trade-off is that it's read-only and public content only (more under Limits).
Every response has the same envelope: the result is in data, and meta reports the charge (meta.billed, meta.charged_usd, meta.empty). Empty results come back as HTTP 200 with meta.empty: true and are not charged. Errors are {"error": {"code", "message", ...}} and are never charged. A post or subreddit that doesn't exist answers 404, so the helper turns that into None. A 429 carries a Retry-After header, so the helper waits and retries once:
import os
import time
import requests
BASE = "https://api.qovia.ai"
HEADERS = {"Authorization": f"Bearer {os.environ['QOVIA_KEY']}"}
spent = 0.0
def get(path, **params):
global spent
params = {k: v for k, v in params.items() if v is not None}
for attempt in range(2):
r = requests.get(f"{BASE}{path}", headers=HEADERS, params=params, timeout=60)
if r.status_code == 429 and attempt == 0:
time.sleep(int(r.headers.get("Retry-After", "10")))
continue
if r.status_code == 404:
return None # missing post or subreddit, not charged
r.raise_for_status()
body = r.json()
spent += float((body.get("meta") or {}).get("charged_usd") or 0)
return body.get("data") or {}
Unknown parameters are rejected with 400 invalid_params, which is why the helper drops None values instead of sending them.
GET /v1/raw/reddit/search takes a keyword (1 to 200 characters) and Reddit's search syntax works inside it: AND, OR, NOT, quoted phrases, subreddit:, author:, title: and so on. Two more parameters matter for monitoring:
sort: relevance (the default), hot, top, new or comments. For monitoring you want time_range: all (the default), year, month, week, day or hour. If you run hourly, Results come about 7 a page. Pass cursor back for the next page and stop when has_more is false. If Reddit spell-corrected your query, corrected_query tells you what it actually searched, which is worth logging: brand names get "corrected" more often than you'd think.
Each post record has text (title and body), author (username), metrics (likes, which is the post's score, and comments) and extra with source_id (the post's fullname, t3_...), title, subreddit, published_at (UTC), flair, content_type, link_url and url.
QUERIES = {
"brand": '"acme analytics" OR acmeanalytics',
"competitor": '"plausible analytics" alternative',
}
def search(query, time_range="day", max_pages=2):
posts, cursor = [], None
for _ in range(max_pages):
page = get("/v1/raw/reddit/search", keyword=query, sort="new",
time_range=time_range, cursor=cursor)
if not page:
break
if page.get("corrected_query"):
print(f" note: Reddit searched {page['corrected_query']!r} instead of {query!r}")
posts.extend(page.get("items", []))
if not page.get("has_more"):
break
cursor = page.get("cursor")
return posts
Quote multi-word names. Without quotes, acme analytics matches any post containing both words anywhere, which for a common word gets noisy fast.
Search finds posts that name you. The more valuable posts often don't: "what do you use for privacy-friendly analytics?" in r/webdev never mentions any product. For those, watch the newest posts in the communities your users hang out in and filter locally.
GET /v1/raw/reddit/subreddit_posts takes a subreddit (the name after r/) and a sort (hot by default; also best, new, top, controversial, rising). It's cursor-paginated: about 6 posts on the first page, then about 24. The list records carry the title and the first 300 characters of the body (extra.truncated tells you when it was cut), which is plenty for a phrase filter. Fields the list doesn't carry (language, author ID, images, upvote ratio) are null; GET /v1/raw/reddit/post has them if you need them.
import re
SUBREDDITS = ["webdev", "SaaS", "selfhosted"]
TOPIC = re.compile(r"analytics|tracking|pageviews", re.I)
INTENT = re.compile(r"alternative to|looking for|recommend|what do you use|switch(ed|ing)? from", re.I)
def intent_posts(subreddit, max_pages=2):
hits, cursor = [], None
for _ in range(max_pages):
page = get("/v1/raw/reddit/subreddit_posts", subreddit=subreddit,
sort="new", cursor=cursor)
if not page:
break # 404: no such subreddit
for p in page.get("items", []):
text = p.get("text") or ""
if TOPIC.search(text) and INTENT.search(text):
hits.append(p)
if not page.get("has_more"):
break
cursor = page.get("cursor")
return hits
The two regexes are deliberately simple: a post must be about the topic and sound like someone is choosing a tool. Tune them on a week of output; this is the part of the script you'll edit most.
Both sources return the same record shape with the same IDs, so one set of seen fullnames is enough. A JSON file next to the script is fine for an hourly job:
import json
from pathlib import Path
STATE = Path("reddit_seen.json")
def load_seen():
return set(json.loads(STATE.read_text())) if STATE.exists() else set()
def save_seen(seen):
STATE.write_text(json.dumps(sorted(seen)))
def only_new(posts, seen):
fresh = []
for p in posts:
pid = p["extra"]["source_id"]
if pid not in seen:
seen.add(pid)
fresh.append(p)
return fresh
This also de-duplicates a post that shows up in both a search and a subreddit feed during the same run.
A title tells you a thread exists; the comments tell you whether it's worth answering. GET /v1/raw/reddit/comments takes the post's post_id (the same t3_... fullname) and an optional sort (best by default, or top, new, old, controversial). A page holds 8 comments: a few top-level ones plus the replies Reddit expands under them, flattened in Reddit's order. Each has cid, text, likes (the comment's score), reply_count, created_at (UTC), pinned, author, depth (0 for top-level) and parent_cid. The page also has total, cursor and has_more. A post without comments answers empty and isn't charged.
def top_comments(post_id, keywords=None):
page = get("/v1/raw/reddit/comments", post_id=post_id, sort="top") or {}
out = []
for c in page.get("items", []):
text = (c.get("text") or "").replace("\n", " ")
out.append({
"cid": c["cid"],
"depth": c.get("depth"),
"score": c.get("likes"),
"replies": c.get("reply_count"),
"mentions": bool(keywords and keywords.search(text)),
"text": text[:300],
})
return out, page.get("total")
One page per post keeps the cost predictable. If you need a full thread, keep paging with cursor; replies Reddit collapses come with a replies_cursor that GET /v1/raw/reddit/comment_replies expands (it takes post_id and that cursor). For alerts, the top 8 is enough to see whether people are already recommending someone.
import csv
from datetime import datetime, timezone
FIELDS = ["checked_at", "source", "post_id", "subreddit", "title", "author",
"score", "comments", "published_at", "url", "top_comment"]
MENTION = re.compile(r"acme", re.I) # flags comments that name you
def now():
return datetime.now(timezone.utc).isoformat(timespec="seconds")
def row(source, p, comments):
ex, m = p.get("extra") or {}, p.get("metrics") or {}
best = comments[0]["text"] if comments else ""
return {
"checked_at": now(),
"source": source,
"post_id": ex.get("source_id"),
"subreddit": ex.get("subreddit"),
"title": ex.get("title"),
"author": (p.get("author") or {}).get("username"),
"score": m.get("likes"),
"comments": m.get("comments"),
"published_at": ex.get("published_at"),
"url": ex.get("url"),
"top_comment": best[:200],
}
def append_csv(path, rows):
path = Path(path)
new_file = not path.exists()
with path.open("a", newline="", encoding="utf-8") as f:
w = csv.DictWriter(f, fieldnames=FIELDS)
if new_file:
w.writeheader()
w.writerows(rows)
def notify(rows):
url = os.environ.get("SLACK_WEBHOOK_URL")
if not url or not rows:
return
lines = [f"*{len(rows)} new Reddit posts*"]
for r in rows[:15]:
lines.append(f"• [{r['source']}] r/{r['subreddit']}: <{r['url']}|{r['title']}> "
f"({r['score']} pts, {r['comments']} comments)")
requests.post(url, json={"text": "\n".join(lines)}, timeout=30)
def run():
seen = load_seen()
found = []
for name, q in QUERIES.items():
found += [(name, p) for p in only_new(search(q), seen)]
for sub in SUBREDDITS:
found += [(f"r/{sub}", p) for p in only_new(intent_posts(sub), seen)]
rows = []
for source, p in found:
comments, _ = top_comments(p["extra"]["source_id"], MENTION)
if any(c["mentions"] for c in comments):
source += " +mentioned in comments"
rows.append(row(source, p, comments))
append_csv("reddit_mentions.csv", rows)
notify(rows)
save_seen(seen)
print(f"{len(rows)} new posts, spent this run: ${spent:.4f}")
if __name__ == "__main__":
run()
The state is saved only after the CSV and the Slack message go out, so a crash mid-run means the next run reports those posts again instead of losing them.
Run it once by hand, then schedule it:
python3 reddit_monitor.py
0 * * * * cd /path/to/monitor && . ./.env && python3 reddit_monitor.py >> monitor.log 2>&1
Output looks something like this (illustrative, not real data):
note: Reddit searched 'acme analytic' instead of 'acmeanalytics'
4 new posts, spent this run: $0.0140
And in Slack (also illustrative):
4 new Reddit posts
• [brand] r/webdev: Anyone tried Acme Analytics for a static site? (12 pts, 9 comments)
• [competitor] r/selfhosted: Plausible alternative that runs on SQLite? (31 pts, 22 comments)
• [r/SaaS] r/SaaS: Looking for privacy-friendly analytics for a B2B app (5 pts, 3 comments)
• [r/webdev +mentioned in comments] r/webdev: What do you use for pageview tracking in 2026? (48 pts, 61 comments)
The last one is the interesting kind: a post that never names you, where someone in the comments already did.
For one-off research ("what did r/selfhosted say about analytics tools this month?") a script is overkill. The same endpoints are available through Qovia's MCP server at https://api.qovia.ai/mcp (Streamable HTTP, same API key in an Authorization: Bearer header, same billing as the HTTP API). In Claude Code:
claude mcp add --transport http qovia https://api.qovia.ai/mcp \
--header "Authorization: Bearer $QOVIA_KEY"
Then ask something like "search Reddit for posts from the last week asking for a Plausible alternative, and summarize what the top comments recommend." Configs for Cursor, VS Code and Claude Desktop are in the MCP docs. The assistant is good for digging into one question; the cron job is better for not missing anything.
Every Reddit endpoint used here (search, subreddit_posts, comments, and post / comment_replies if you add them) is a fixed $0.001 per successful call ($1 per 1,000). Empty results, errors and 404s are not charged, so a quiet hour where search finds nothing costs nothing for that query.
As a rough example (arithmetic only): with 2 queries x up to 2 search pages, 3 subreddits x up to 2 pages, and one comment page per new post, a run is at most 10 calls plus one per new post. With 4 new posts that's $0.014. Run hourly, the worst case is 240 calls a day ($0.24) plus one call per new post, so roughly $7 to $8 a month; it's less in practice because a search with no new results returns empty and isn't charged. The subreddit feeds are the fixed part: one page per subreddit, or running every three hours, cuts that by half or more. The spent counter prints the real number each run. New accounts get 50 trial calls, counted only when a call returns data; after that it's a prepaid balance with no monthly fee. Details are on the pricing page, and every parameter is in the API docs.
The default rate limit is 60 requests per minute per account, far more than an hourly job needs.
subreddit_posts carries only the first 300 characters of the body; fetch GET /v1/raw/reddit/post when you need the full text.corrected_query and test your queries on reddit.com once before trusting them.
If you build on this, tell me in the comments what you'd want it to flag besides mentions.
Drafted with AI assistance and checked against the live API docs.