How to Monitor Donald Trump Tweets in Real Time with TwexAPI

If your team watches public posts from @realDonaldTrump, manual refreshing is a brittle workflow. Someone has to keep checking X, copy links into a chat, and decide whether the post is new, duplicated, or already handled.

TwexAPI's Advanced Twitter Search by Page endpoint can be used as a near-real-time monitor. Poll from:realDonaldTrump -filter:retweets with sortBy: "Latest", store every tweet_id you have already seen, and alert only when a new public post appears after the monitor is running.

This is a polling workflow, not a streaming socket. The reliability comes from state: baseline the first page, persist seen IDs, save raw posts before alerting, and make duplicate alerts impossible.

When This Workflow Fits

Use this workflow when you need a lightweight monitor for new public Trump posts on X.

Common use cases include:

• Newsroom, research, or analyst alerts when a new public post appears.
• Internal monitoring around policy, election, market, or media narratives.
• A watchlist that routes new posts into Slack, email, Telegram, or a webhook.
• A durable record of when your system first observed each public post.

Do not treat this as a complete historical archive. For backfills and deeper collection, use the paginated export pattern and continue with next_cursor until your target window is covered.

API Endpoint

Send a POST request to:

https://api.twexapi.io/twitter/advanced_search/page

The request body for a monitor is intentionally small:

Each page returns up to 20 tweets with data, has_next_page, and next_cursor. For real-time monitoring, poll the first latest page repeatedly and deduplicate by tweet_id.

curl --request POST \
  --url https://api.twexapi.io/twitter/advanced_search/page \
  --header 'Authorization: Bearer <token>' \
  --header 'Content-Type: application/json' \
  --data '{
    "searchTerms": [
      "from:realDonaldTrump -filter:retweets"
    ],
    "sortBy": "Latest"
  }'

A shortened response looks like this:

1{
2  "code": 200,
3  "msg": "success",
4  "data": [
5    {
6      "tweet_id": "1234567890123456789",
7      "created_at": "Mon Jul 20 12:34:56 +0000 2026",
8      "text": "<post text returned by the API>",
9      "favorite_count": 1000,
10      "retweet_count": 200,
11      "reply_count": 300
12    }
13  ],
14  "has_next_page": true,
15  "next_cursor": "cursor_from_this_response"
16}

Monitoring Model

A dependable monitor should separate collection, state, storage, and alerting:

1. Fetch the latest page for from:realDonaldTrump -filter:retweets.
2. If there is no saved state, seed the current tweet_id values without alerting.
3. On later runs, compare returned IDs with the stored seen_ids.
4. Save each new raw post to durable storage before sending alerts.
5. Send one alert per new tweet_id, then persist the updated state.
6. Sleep for a reasonable interval, such as 1 to 5 minutes for high-priority monitoring.

The first run is important. If you alert on everything returned by the first page, the monitor may notify old posts as if they just happened.

Python Monitor with Durable State

This script stores seen IDs in a local JSON file and writes every new post to JSONL before sending an optional webhook alert. For production, replace the local files with a database, Redis, S3, or your team's normal event store.

1import json
2import os
3import time
4from datetime import datetime, timezone
5from pathlib import Path
6from typing import Any
7
8import requests
9
10TOKEN = os.environ["TWEXAPI_BEARER_TOKEN"]
11WEBHOOK_URL = os.environ.get("ALERT_WEBHOOK_URL")
12
13API_URL = "https://api.twexapi.io/twitter/advanced_search/page"
14QUERY = "from:realDonaldTrump -filter:retweets"
15SORT_BY = "Latest"
16POLL_SECONDS = int(os.environ.get("POLL_SECONDS", "180"))
17
18STATE_PATH = Path("trump-monitor-state.json")
19EVENTS_PATH = Path("trump-post-events.jsonl")
20
21HEADERS = {
22    "Authorization": f"Bearer {TOKEN}",
23    "Content-Type": "application/json",
24}
25
26def load_seen_ids() -> set[str]:
27    if not STATE_PATH.exists():
28        return set()
29    state = json.loads(STATE_PATH.read_text(encoding="utf-8"))
30    return set(state.get("seen_ids") or [])
31
32def save_seen_ids(seen_ids: set[str]) -> None:
33    STATE_PATH.write_text(
34        json.dumps(
35            {
36                "query": QUERY,
37                "sortBy": SORT_BY,
38                "updated_at": datetime.now(timezone.utc).isoformat(),
39                "seen_ids": sorted(seen_ids),
40            },
41            ensure_ascii=False,
42            indent=2,
43        ),
44        encoding="utf-8",
45    )
46
47def fetch_latest_posts() -> list[dict[str, Any]]:
48    response = requests.post(
49        API_URL,
50        headers=HEADERS,
51        json={"searchTerms": [QUERY], "sortBy": SORT_BY},
52        timeout=30,
53    )
54    response.raise_for_status()
55    body = response.json()
56    return [item for item in body.get("data", []) if item]
57
58def post_url(tweet_id: str) -> str:
59    return f"https://x.com/realDonaldTrump/status/{tweet_id}"
60
61def store_event(tweet: dict[str, Any]) -> None:
62    event = {
63        "observed_at": datetime.now(timezone.utc).isoformat(),
64        "query": QUERY,
65        "tweet": tweet,
66    }
67    with EVENTS_PATH.open("a", encoding="utf-8") as f:
68        f.write(json.dumps(event, ensure_ascii=False) + "\n")
69
70def send_alert(tweet: dict[str, Any]) -> None:
71    tweet_id = tweet.get("tweet_id")
72    text = tweet.get("full_text") or tweet.get("text") or ""
73    payload = {
74        "type": "new_trump_post",
75        "tweet_id": tweet_id,
76        "url": post_url(tweet_id),
77        "created_at": tweet.get("created_at_datetime") or tweet.get("created_at"),
78        "text": text,
79    }
80
81    if not WEBHOOK_URL:
82        print("NEW", payload["url"], text[:180])
83        return
84
85    response = requests.post(WEBHOOK_URL, json=payload, timeout=15)
86    response.raise_for_status()
87
88def monitor_once(seen_ids: set[str], initialized: bool) -> bool:
89    latest = fetch_latest_posts()
90
91    if not initialized:
92        for tweet in latest:
93            tweet_id = tweet.get("tweet_id")
94            if tweet_id:
95                seen_ids.add(tweet_id)
96        save_seen_ids(seen_ids)
97        print(f"Initialized baseline with {len(seen_ids)} seen posts")
98        return True
99
100    new_posts = []
101    for tweet in latest:
102        tweet_id = tweet.get("tweet_id")
103        if tweet_id and tweet_id not in seen_ids:
104            new_posts.append(tweet)
105
106    for tweet in reversed(new_posts):
107        tweet_id = tweet["tweet_id"]
108        store_event(tweet)
109        send_alert(tweet)
110        seen_ids.add(tweet_id)
111
112    if new_posts:
113        save_seen_ids(seen_ids)
114
115    print(f"Checked latest page; new posts: {len(new_posts)}")
116    return initialized
117
118def main() -> None:
119    seen_ids = load_seen_ids()
120    initialized = STATE_PATH.exists()
121
122    while True:
123        try:
124            initialized = monitor_once(seen_ids, initialized)
125        except Exception as exc:
126            print(f"Monitor error: {exc}")
127        time.sleep(POLL_SECONDS)
128
129if __name__ == "__main__":
130    main()

Run it with:

export TWEXAPI_BEARER_TOKEN="<token>"
export ALERT_WEBHOOK_URL="https://example.com/webhook"
export POLL_SECONDS="180"
python monitor_trump_posts.py

If ALERT_WEBHOOK_URL is not set, the script prints alerts to stdout. That makes the first dry run easier to inspect.

Alert Payload Design

Keep alerts small. Downstream systems should receive enough context to route the event, then use the stored raw payload for deeper analysis.

{
  "type": "new_trump_post",
  "tweet_id": "1234567890123456789",
  "url": "https://x.com/realDonaldTrump/status/1234567890123456789",
  "created_at": "Mon Jul 20 12:34:56 +0000 2026",
  "text": "<post text returned by the API>"
}

Good alert rules are boring:

• One alert per tweet_id.
• Store the raw post before sending the alert.
• Include the query that found the post.
• Retry failed webhook delivery without creating duplicate events.
• Keep human review in the workflow for interpretation.

Polling Frequency

For high-priority monitoring, start with a 1 to 5 minute interval. A shorter interval gives fresher alerts but costs more API calls and creates more operational noise. A longer interval is usually enough for research, daily summaries, and non-urgent dashboards.

Use this as a starting point:

Avoid making the interval so short that retries overlap. One running monitor with clean state is better than several competing monitors sending duplicate alerts.

When to Use Pagination

For a live monitor, repeatedly fetching the latest page is usually enough. Use next_cursor when you need to catch up after downtime or backfill older results.

A catch-up job can:

1. Fetch the latest page.
2. Alert on new IDs that are not in seen_ids.
3. Follow next_cursor for a limited number of pages.
4. Stop when every item on a page is already known or when the page limit is reached.

Keep catch-up separate from the live loop. That makes failures easier to reason about and avoids turning a real-time monitor into an accidental historical crawler.

Common Pitfalls

• Do not call this a true streaming integration. It is polling against the latest search page.
• Do not alert on the first run unless you intentionally want a backlog notification.
• Do not change searchTerms without resetting or versioning your state file.
• Do not use Top for real-time monitoring. Use Latest.
• Do not rely on in-memory seen_ids for production. Persist state before the process exits.
• Do not interpret a public figure's post without context. The monitor detects new public posts; humans still decide significance.

Related Resources

• Advanced Search by Page API Reference
• How to Collect Donald Trump X Posts with Advanced Search
• How to Monitor New Public Serenity Posts on X

Wrap-up

To monitor Trump tweets with TwexAPI, poll POST /twitter/advanced_search/page with from:realDonaldTrump -filter:retweets, sort by Latest, and keep durable tweet_id state. The first run establishes a baseline; every later run compares the latest page against that baseline and alerts only on new public posts.

That pattern gives you near-real-time visibility without duplicate notifications or unclear collection state.