Threads Keyword Search API in Python: Monitor Brand Mentions

The dataset fits on a card: three phrases. Product name, the misspelling reviewers already catch, this quarter's campaign tag. That is the whole watchlist — not a synonym taxonomy, not a customer roster, not an Instagram pagination_token. search-posts on the Threads Public Data API takes one required query. This recipe's job is to ship threads_matches.csv: one row per nested thread_items post, stamped with the phrase that hit and the UTC time you observed it. A keyword hit is a public post in a sample. It is not a complaint, not a customer, and not proof you have every mention.

Sign in and copy an authorized platform key from the dashboard. The script reads OPENDATA_API_KEY from the server environment. Leave YOUR_API_KEY as a placeholder in any URL you paste. Opening this page or copying the script does not send a request. A call happens only when you replace the placeholder and run it. The key rides on the query string, so keep production traffic server-side and redact URLs from logs, tickets, and screenshots. You do not need a Threads login.

Three phrases are the whole watchlist

A brand-review queue does not start at a thousand synonyms. Three phrases you already read by hand are enough to produce the file. Extra phrases spend credits on near-duplicate samples and bury the reviewer in restatements of the same thread. This page runs the documented public example coffee so copying the snippet is not a live search against your brand list. Put the other two phrases in the same Python list when you intend to spend credits. Neither coffee nor a phrase you add is a guaranteed hit, and a hit is not a reputation score.

Each phrase is its own GET to https://api.opendata-api.com/api/project/threads-scraper/search-posts. Only query is accepted. Advanced operators, date filters, sort modes, page, and pagination_token are not documented. Do not invent them. Python params URL-encodes spaces and punctuation; do not concatenate the query string by hand. Chinese or punctuated phrases still belong in query. They are not extra parameters. Three phrases are three calls, not one request carrying an array.

Transport failure, HTTP 4xx/5xx, and a gateway envelope that is not business-success are three different stops. The envelope is {code, message, data}. HTTP 2xx means bytes arrived. Gateway code 200 means the platform accepted the call. Inside data, the operation payload still has its own status (observed as ok on success), request_id, and nested data. HTTP 200 is not a mention. Gateway 200 is not a mention. After those checks, an empty results array is a valid sample of nothing — not an outage to paper over with a fake zero, and not a reason to retry until the spreadsheet looks busy. One empty phrase beside another phrase with rows is still a finished run.

Unwrap thread_items, keep the post id

Individual posts are not the outer search wrappers. After both status layers succeed, walk data.results[].node.thread.thread_items[].post on the operation payload. One wrapper can hold several posts: a parent and a follow-up in the same thread can both contain the phrase while the outer array still counts one object. Counting len(results) as the mention count undercounts. Skipping thread_items drops posts sitting in the same thread. This is not Instagram user-post pagination and not a cursor walk. There is no token to copy forward.

From each post, keep pk, id, and shortcode code as separate strings. Never coerce them to floating-point numbers. Collection id can carry an account suffix; it is not interchangeable with code, and it is not the selector for get-post-details. That operation wants code or an HTTPS post url on threads.com or threads.net, and this recipe does not call it. The CSV is the review queue. Enrich a shortcode later only when a reviewer points at a row.

caption.text and user.username are attribution when present. taken_at is Unix publication time in seconds. Empty text is empty, not a reason to drop the row — a reviewer still needs the id. A match on coffee is not a complaint, not sentiment, and not permission to contact the author. Relevance stays in your process. The same post hitting two phrases is two query associations, not two brands.

The CSV a reviewer can open

Install requests. Run the script on the server. It writes threads_matches.csv in the working directory: append-safe, first observation wins on string pk, later duplicate ids from the same or another phrase are skipped. Columns a reviewer can sort without breaking identifiers:

pk, id, code, username, text, taken_at, like_count, query, observed_at_utc, request_id

like_count is an observation at retrieval time, not reach. Leave missing cells empty; do not write 0 for unknown. Import identifier and text columns as text in a spreadsheet so long ids do not become scientific notation and untrusted captions do not become formulas. Prefix cells that start with =, +, -, or @ if someone will open the file in Excel.

observed_at_utc is when you retrieved the sample. It is not taken_at. Keep both. Persist request_id for redacted diagnostics; do not log the key.

import csv
import os
from datetime import datetime, timezone
from pathlib import Path

import requests

BASE = "https://api.opendata-api.com/api/project/threads-scraper"
API_KEY = os.environ["OPENDATA_API_KEY"]
# Documented public example. Add the other two phrases locally when you mean to spend credits.
WATCHLIST = ["coffee"]
OUTPUT = Path("threads_matches.csv")
FIELDS = [
    "pk", "id", "code", "username", "text", "taken_at",
    "like_count", "query", "observed_at_utc", "request_id",
]


def read_search(query):
    try:
        response = requests.get(
            f"{BASE}/search-posts",
            params={"key": API_KEY, "query": query},
            timeout=30,
        )
    except requests.RequestException:
        raise RuntimeError("Transport failure; inspect redacted logs") from None
    if not 200 <= response.status_code < 300:
        raise RuntimeError(f"HTTP failure: {response.status_code}")
    try:
        envelope = response.json()
    except ValueError:
        raise RuntimeError("Expected a JSON response") from None
    if not isinstance(envelope, dict) or str(envelope.get("code")) != "200":
        raise RuntimeError("Gateway business failure")
    payload = envelope.get("data")
    if not isinstance(payload, dict) or payload.get("status") != "ok":
        raise RuntimeError("Operation failed or has an unexpected status")
    nested = payload.get("data")
    if not isinstance(nested, dict) or not isinstance(nested.get("results"), list):
        raise RuntimeError("Expected a search results array")
    return payload, nested["results"]


def existing_pks(path):
    if not path.exists() or path.stat().st_size == 0:
        return set()
    with path.open(newline="", encoding="utf-8") as source:
        return {row["pk"] for row in csv.DictReader(source) if row.get("pk")}


def flatten(results):
    for result in results:
        try:
            items = result["node"]["thread"]["thread_items"]
        except (TypeError, KeyError):
            raise RuntimeError("Unexpected search wrapper") from None
        if not isinstance(items, list):
            raise RuntimeError("Expected thread_items to be a list")
        for item in items:
            try:
                post = item["post"]
            except (TypeError, KeyError):
                raise RuntimeError("Missing post on thread item") from None
            yield post


def rows_for(query, payload, results, seen):
    observed_at = datetime.now(timezone.utc).isoformat()
    new_rows = []
    for post in flatten(results):
        pk = post.get("pk")
        if not isinstance(pk, str) or not pk:
            raise RuntimeError("Expected a string post pk")
        if pk in seen:
            continue
        seen.add(pk)
        new_rows.append({
            "pk": pk,
            "id": post.get("id"),
            "code": post.get("code"),
            "username": (post.get("user") or {}).get("username"),
            "text": (post.get("caption") or {}).get("text"),
            "taken_at": post.get("taken_at"),
            "like_count": post.get("like_count"),
            "query": query,
            "observed_at_utc": observed_at,
            "request_id": payload.get("request_id"),
        })
    return new_rows


if __name__ == "__main__":
    seen = existing_pks(OUTPUT)
    write_header = not OUTPUT.exists() or OUTPUT.stat().st_size == 0
    with OUTPUT.open("a", newline="", encoding="utf-8") as target:
        writer = csv.DictWriter(target, fieldnames=FIELDS)
        if write_header:
            writer.writeheader()
        for query in WATCHLIST:
            payload, results = read_search(query)
            writer.writerows(rows_for(query, payload, results, seen))

This single-process file is not a database. Concurrent workers need a unique constraint on the string post key. If you monitor several phrases, store the query-to-post link separately so one post that matched twice is still one post.

Search is not a complete mention archive

data.cursor may appear. It does not authorize a follow-up parameter. search-posts accepts only query. list-user-posts is a different sample — path data.posts[].node.thread_items[].post, input username — and also has no pagination input. An account timeline answers what is visible on that handle. Keyword search answers what this phrase hit this time. Neither is a complete history. Polling faster does not prove coverage; it repeats the same window sooner. The 30-second timeout in the script is a client choice, not a latency promise.

Budget credits per phrase per run. Three phrases are three searches. Do not call post-detail on every row. Unavailable profiles, empty user-search arrays, and a null view_count belong to other operations; they are not reasons to invent zeros in this CSV. Schedule bounded runs yourself. Respect platform terms, lawful processing, data minimization, and retention.

A timestamped CSV of nested thread items is a working copy of a sample. It is not an archive of every public mention. Start from the Threads API reference.