> ## Documentation Index
> Fetch the complete documentation index at: https://docs.sorsa.io/llms.txt
> Use this file to discover all available pages before exploring further.

# Real-Time Monitoring

# Real-Time Monitoring

Detect new tweets from specific accounts, track keyword mentions as they happen, and feed live X data into your application. This guide shows how to build a near-real-time monitoring pipeline on top of the Sorsa API using a pull-based polling pattern.

Sorsa returns fresh data on every request. If a tweet was posted half a second ago, the next API call will include it. Combined with a 20 requests-per-second rate limit and response times around 300ms, polling-based loops match or beat most managed streaming services for the use cases below, without persistent connections, reconnect logic, or OAuth.

> **Free to prototype:** Every Sorsa endpoint is available on your first 100 requests: one-time, no credit card, no expiry. That is enough to stand up any of the monitors below, confirm it detects live tweets, and validate your Slack or Discord routing before choosing a plan. Because polling is request-heavy, size a paid plan to your poll interval using the usage table further down.

> **Note:** For a fuller walkthrough with additional architecture patterns and end-to-end examples, see [Real-Time Twitter Monitoring with REST API](https://api.sorsa.io/blog/real-time-twitter-monitoring) on the blog.

***

## How Polling-Based Monitoring Works

There are two ways to get data from a social platform: push-based (streaming, webhooks) or pull-based (polling). The Sorsa API uses polling. The pattern has four steps:

1. **Poll an endpoint** at a regular interval (1 to 30 seconds).
2. **Compare results** against previously seen tweet IDs to identify what is new.
3. **Process new tweets:** send alerts, store, fan out to Slack, Discord, etc.
4. **Repeat.**

Tweet IDs (Snowflake) are monotonically increasing, so deduplication is reliable: a higher ID always means a newer tweet. If a script crashes, it resumes from the last checkpoint on the next poll. No persistent connection to manage, no reconnect strategy required.

### Choosing the Right Endpoint

| What you want to monitor     | Endpoint         | Method | Why this one                                                              |
| :--------------------------- | :--------------- | :----- | :------------------------------------------------------------------------ |
| A single account             | `/user-tweets`   | POST   | Returns the latest tweets from one user's timeline                        |
| Up to 5,000 accounts at once | `/list-tweets`   | GET    | One request covers all members of an X List                               |
| A keyword or hashtag         | `/search-tweets` | POST   | Full search operator support, chronological ordering with `order: latest` |
| @mentions of an account      | `/mentions`      | POST   | Purpose-built for mention tracking with engagement filters                |

***

## Level 1: Monitor a Single Account

The simplest case. The loop polls `/user-tweets` and emits tweets newer than the last seen ID.

### Python

```python theme={null}
import requests
import time

API_KEY = "YOUR_API_KEY"
USERNAME = "elonmusk"
POLL_INTERVAL = 5  # seconds

URL = "https://api.sorsa.io/v3/user-tweets"
HEADERS = {"ApiKey": API_KEY, "Content-Type": "application/json"}

last_seen_id = None

print(f"Monitoring @{USERNAME}...")

while True:
    try:
        resp = requests.post(URL, headers=HEADERS, json={"username": USERNAME})
        resp.raise_for_status()
        tweets = resp.json().get("tweets", [])

        if tweets:
            # Snowflake IDs arrive as strings. Use the highest (newest) ID in the
            # batch; this stays correct even if a pinned tweet appears first.
            top_id = max(int(t["id"]) for t in tweets)

            if last_seen_id is None:
                last_seen_id = top_id
                print(f"Baseline set: {last_seen_id}")
            else:
                new_tweets = [t for t in tweets if int(t["id"]) > last_seen_id]
                for tweet in reversed(new_tweets):  # oldest first
                    print(f"[NEW] @{USERNAME}: {tweet['full_text'][:140]}")
                if new_tweets:
                    last_seen_id = top_id

    except requests.exceptions.RequestException as e:
        print(f"Error: {e}")
        time.sleep(POLL_INTERVAL * 2)
        continue

    time.sleep(POLL_INTERVAL)
```

### JavaScript

```javascript theme={null}
const API_KEY = "YOUR_API_KEY";
const USERNAME = "elonmusk";
const POLL_INTERVAL = 5000;

let lastSeenId = null;
console.log(`Monitoring @${USERNAME}...`);

while (true) {
  try {
    const resp = await fetch("https://api.sorsa.io/v3/user-tweets", {
      method: "POST",
      headers: { "ApiKey": API_KEY, "Content-Type": "application/json" },
      body: JSON.stringify({ username: USERNAME }),
    });
    if (!resp.ok) throw new Error(`HTTP ${resp.status}`);

    const tweets = (await resp.json()).tweets || [];

    if (tweets.length > 0) {
      // BigInt avoids precision loss on 64-bit Snowflake IDs.
      // Use the highest ID in the batch (robust if a pinned tweet appears first).
      let topId = 0n;
      for (const t of tweets) {
        const id = BigInt(t.id);
        if (id > topId) topId = id;
      }

      if (lastSeenId === null) {
        lastSeenId = topId;
        console.log(`Baseline set: ${lastSeenId}`);
      } else {
        const newTweets = tweets.filter((t) => BigInt(t.id) > lastSeenId);
        for (const t of [...newTweets].reverse()) {
          console.log(`[NEW] @${USERNAME}: ${t.full_text.slice(0, 140)}`);
        }
        if (newTweets.length) lastSeenId = topId;
      }
    }
  } catch (err) {
    console.error(`Error: ${err.message}`);
    await new Promise((r) => setTimeout(r, POLL_INTERVAL * 2));
    continue;
  }
  await new Promise((r) => setTimeout(r, POLL_INTERVAL));
}
```

This scales poorly for many accounts. Monitoring 50 accounts means 50 separate loops and 50x the API requests. That is where X Lists come in.

***

## Level 2: Monitor Many Accounts with a Single Request

X Lists group up to 5,000 accounts. `/list-tweets` returns the merged latest tweets across all members in one API call. This is the default pattern for production multi-account monitoring. See [Lists & Communities](https://docs.sorsa.io/lists-and-communities) for more.

### Step 1: Create a Public X List

1. Go to [X Lists](https://x.com/i/lists) and create a list.
2. Add the accounts to monitor (up to 5,000).
3. Set the list to **Public**. Private lists are not accessible via the API.
4. Copy the **List ID** from the URL. For `https://x.com/i/lists/1234567890` the ID is `1234567890`.

### Step 2: Poll the List

```python theme={null}
import requests
import time

API_KEY = "YOUR_API_KEY"
LIST_ID = "YOUR_LIST_ID"
POLL_INTERVAL = 5

URL = f"https://api.sorsa.io/v3/list-tweets?list_id={LIST_ID}"
HEADERS = {"ApiKey": API_KEY, "Accept": "application/json"}


def monitor_list(callback, interval=POLL_INTERVAL):
    """Poll an X List and call `callback` for each new tweet detected."""
    last_seen_id = None
    print(f"Monitoring List {LIST_ID} (interval: {interval}s)")

    while True:
        try:
            resp = requests.get(URL, headers=HEADERS, timeout=10)
            resp.raise_for_status()
            tweets = resp.json().get("tweets", [])

            if not tweets:
                time.sleep(interval)
                continue

            top_id = max(int(t["id"]) for t in tweets)

            if last_seen_id is None:
                last_seen_id = top_id
                print(f"Baseline set: {last_seen_id}")
            else:
                new_tweets = [t for t in tweets if int(t["id"]) > last_seen_id]
                if new_tweets:
                    for tweet in reversed(new_tweets):
                        callback(tweet)
                    last_seen_id = top_id

        except requests.exceptions.RequestException as e:
            print(f"Request error: {e}. Retrying in {interval * 2}s")
            time.sleep(interval * 2)
            continue

        time.sleep(interval)


def on_new_tweet(tweet):
    user = tweet["user"]
    print(f"[NEW] @{user['username']}: {tweet['full_text'][:120]}")
    print(
        f"       Likes: {tweet.get('likes_count', 0)} | "
        f"RTs: {tweet.get('retweet_count', 0)} | "
        f"Views: {tweet.get('view_count', 'N/A')}\n"
    )


if __name__ == "__main__":
    monitor_list(on_new_tweet)
```

**Efficiency gain.** Polling 50 accounts individually at a 10-second interval costs 50 x 8,640 = 432,000 requests per day. The same 50 accounts in one List polled at 10 seconds costs 8,640 requests per day. A 50x reduction. For more patterns like this, see [Optimizing API Usage](https://docs.sorsa.io/optimizing-api-usage).

> `/list-tweets` returns up to 20 tweets per page. If list members tweet faster than that within one poll interval, drop the interval to 2 to 3 seconds, or paginate via `next_cursor` until reaching a previously seen ID.

***

## Level 3: Monitor a Keyword or Hashtag

Instead of tracking accounts, poll `/search-tweets` with `order: "latest"` for chronological results matching a query.

```python theme={null}
import requests
import time

API_KEY = "YOUR_API_KEY"
QUERY = '"your brand" OR @yourbrand lang:en'
POLL_INTERVAL = 10

URL = "https://api.sorsa.io/v3/search-tweets"
HEADERS = {"ApiKey": API_KEY, "Content-Type": "application/json"}


def monitor_keyword(query, callback, interval=10):
    last_seen_id = None
    print(f"Monitoring: {query} (interval: {interval}s)")

    while True:
        try:
            resp = requests.post(
                URL,
                headers=HEADERS,
                json={"query": query, "order": "latest"},
                timeout=10,
            )
            resp.raise_for_status()
            tweets = resp.json().get("tweets", [])

            if tweets:
                top_id = max(int(t["id"]) for t in tweets)
                if last_seen_id is None:
                    last_seen_id = top_id
                    print(f"Baseline set: {last_seen_id}")
                else:
                    new_tweets = [t for t in tweets if int(t["id"]) > last_seen_id]
                    for tweet in reversed(new_tweets):
                        callback(tweet)
                    if new_tweets:
                        last_seen_id = top_id

        except requests.exceptions.RequestException as e:
            print(f"Error: {e}")
            time.sleep(interval * 2)
            continue

        time.sleep(interval)


monitor_keyword(QUERY, on_new_tweet, interval=10)
```

Any [search operator](https://docs.sorsa.io/search-operators) works in the query string. To monitor high-engagement English mentions of your brand and exclude retweets:

```python theme={null}
monitor_keyword('"your brand" min_faves:10 lang:en -filter:retweets', on_new_tweet)
```

***

## Routing New Tweets to Slack, Discord, or Any HTTP Endpoint

The polling loop is the producer; the callback decides what happens to each new tweet. Because the callback is just a function, the same monitor can route to anything that speaks HTTP.

### Slack via Incoming Webhook

```python theme={null}
import requests

SLACK_WEBHOOK_URL = "https://hooks.slack.com/services/YOUR/SLACK/WEBHOOK"


def send_to_slack(tweet):
    user = tweet["user"]
    text = (
        f"*New tweet from @{user['username']}*\n"
        f"{tweet['full_text']}\n"
        f"Likes: {tweet.get('likes_count', 0)} | "
        f"RTs: {tweet.get('retweet_count', 0)} | "
        f"Views: {tweet.get('view_count', 'N/A')}\n"
        f"https://x.com/{user['username']}/status/{tweet['id']}"
    )
    requests.post(SLACK_WEBHOOK_URL, json={"text": text})


# Plug into any monitor:
monitor_list(send_to_slack)
# or: monitor_keyword("bitcoin lang:en min_faves:50", send_to_slack)
```

### Discord

```python theme={null}
DISCORD_WEBHOOK_URL = "https://discord.com/api/webhooks/YOUR/WEBHOOK"


def send_to_discord(tweet):
    user = tweet["user"]
    content = (
        f"**@{user['username']}** just tweeted:\n"
        f"{tweet['full_text']}\n"
        f"https://x.com/{user['username']}/status/{tweet['id']}"
    )
    requests.post(DISCORD_WEBHOOK_URL, json={"content": content})
```

### Telegram

```python theme={null}
TELEGRAM_BOT_TOKEN = "YOUR_BOT_TOKEN"
TELEGRAM_CHAT_ID = "YOUR_CHAT_ID"


def send_to_telegram(tweet):
    user = tweet["user"]
    text = (
        f"New tweet from @{user['username']}\n\n"
        f"{tweet['full_text']}\n\n"
        f"https://x.com/{user['username']}/status/{tweet['id']}"
    )
    requests.post(
        f"https://api.telegram.org/bot{TELEGRAM_BOT_TOKEN}/sendMessage",
        json={"chat_id": TELEGRAM_CHAT_ID, "text": text},
    )
```

### Any Custom HTTP Endpoint

```python theme={null}
def send_to_internal_api(tweet):
    requests.post(
        "https://internal.example.com/events/twitter",
        json={
            "tweet_id": tweet["id"],
            "username": tweet["user"]["username"],
            "text": tweet["full_text"],
            "metrics": {
                "likes": tweet.get("likes_count", 0),
                "retweets": tweet.get("retweet_count", 0),
                "views": tweet.get("view_count", 0),
            },
            "url": f"https://x.com/{tweet['user']['username']}/status/{tweet['id']}",
        },
        headers={"Authorization": "Bearer YOUR_INTERNAL_TOKEN"},
        timeout=5,
    )
```

***

## API Usage Calculator

Polling uses one request per cycle. Pick an interval that balances latency against monthly request volume.

| Interval   | Req/Hour | Req/Day | Req/Month (30d) |
| :--------- | :------- | :------ | :-------------- |
| 1 second   | 3,600    | 86,400  | 2,592,000       |
| 5 seconds  | 720      | 17,280  | 518,400         |
| 10 seconds | 360      | 8,640   | 259,200         |
| 30 seconds | 120      | 2,880   | 86,400          |
| 1 minute   | 60       | 1,440   | 43,200          |

For most social listening and brand monitoring, 10 to 30 seconds is sufficient: any new tweet is detected within half a minute. Reserve 1 to 5 second intervals for financial signal detection or breaking-news pipelines.

The free 100 requests are enough to prototype and validate a monitor end to end. For sustained monitoring, match a plan to the monthly volume in the table above: a single loop at 30 to 60 seconds fits within Pro (100,000 requests per month), and a single loop at 10 seconds fits within Enterprise (500,000 requests per month). These figures are per monitor, so running several in parallel multiplies the total; size your plan to the combined volume. See [Pricing](https://api.sorsa.io/pricing) for full plan details.

> For rate limits or volumes above the standard plans, [talk to sales](https://api.sorsa.io/talk-to-sales) for a custom quota, or ask in [Discord](https://discord.com/invite/uwAefKCj7X).

***

## Production Hardening

The examples above work for development. For production, address these five concerns.

### 1. Persist `last_seen_id` Across Restarts

If the script crashes and restarts without remembering its checkpoint, it either reprocesses old tweets (duplicate alerts) or silently skips the gap. Store the last seen ID in a file, database, or Redis.

```python theme={null}
import json
import os

STATE_FILE = "monitor_state.json"


def load_state():
    if os.path.exists(STATE_FILE):
        with open(STATE_FILE) as f:
            return json.load(f).get("last_seen_id")
    return None


def save_state(last_seen_id):
    with open(STATE_FILE, "w") as f:
        json.dump({"last_seen_id": last_seen_id}, f)
```

Load on startup, save after every successful poll that updates the cursor.

### 2. Exponential Backoff for Errors

Network issues, rate limits (HTTP 429), and transient API errors will happen. Back off gradually with a cap rather than retrying immediately. See [Error Codes](https://docs.sorsa.io/error-codes) for the full reference.

```python theme={null}
retry_delay = POLL_INTERVAL
MAX_DELAY = 60

while True:
    try:
        resp = requests.get(URL, headers=HEADERS, timeout=10)
        if resp.status_code == 429:
            print(f"Rate limited. Backing off {retry_delay}s")
            time.sleep(retry_delay)
            retry_delay = min(retry_delay * 2, MAX_DELAY)
            continue
        resp.raise_for_status()
        retry_delay = POLL_INTERVAL  # reset on success
        # process tweets
    except requests.exceptions.RequestException as e:
        print(f"Error: {e}")
        time.sleep(retry_delay)
        retry_delay = min(retry_delay * 2, MAX_DELAY)
        continue

    time.sleep(POLL_INTERVAL)
```

### 3. Separate Polling from Processing

Do not run expensive operations (NLP, database writes, external API calls) synchronously inside the polling loop. If a downstream system slows down, the loop falls behind schedule. Push new tweets into a queue and process them in a separate worker.

```python theme={null}
from collections import deque
import threading

tweet_queue = deque()


def polling_loop():
    """Fast loop: poll and enqueue. No heavy work here."""
    # Standard polling code, but instead of calling callback(tweet):
    # tweet_queue.append(tweet)
    pass


def processing_worker():
    """Separate thread: dequeue and dispatch."""
    while True:
        if tweet_queue:
            tweet = tweet_queue.popleft()
            send_to_slack(tweet)
            save_to_database(tweet)
        else:
            time.sleep(0.1)


threading.Thread(target=processing_worker, daemon=True).start()
polling_loop()
```

For heavier workloads, replace the in-memory deque with Redis, RabbitMQ, SQS, or any message broker the stack already uses.

### 4. Monitor the Monitor

Log each poll cycle: timestamp, new tweet count, response time, errors. Alert if the monitor has not completed a successful poll in the last N minutes; silent failures cause invisible data gaps in alerting pipelines. Operational status of the API is available at the [Sorsa Status Page](https://uptime.sorsa.io/status/v3).

### 5. Handle Edge Cases

* **Deleted tweets:** if a tweet is deleted between fetch and callback, the URL will 404. Treat as expected.
* **Protected accounts:** if a tracked user goes private, `/user-tweets` returns an empty list. Log and continue.
* **Pinned tweets:** the first tweet in a `/user-tweets` response is often the pinned one, not the most recent. Do not use `tweets[0]` as the newest ID; take `max(int(t["id"]) for t in tweets)` instead (as the examples above do), or sort by `created_at`.
* **Retweets:** `tweet["retweeted_status"]` is populated for retweets. Decide whether to include them or filter out.
* **Reply restrictions:** `is_replies_limited` indicates the author restricted replies; useful signal for some monitoring use cases.

***

## Next Steps

* [Search Operators](https://docs.sorsa.io/search-operators): advanced filters to reduce noise in keyword-based monitoring.
* [Track Mentions](https://docs.sorsa.io/search-mentions): dedicated endpoint for @mentions with engagement filters.
* [Rate Limits](https://docs.sorsa.io/rate-limits): handling 429 errors and request patterns.
* [Pagination](https://docs.sorsa.io/pagination): backfill historical data alongside real-time monitoring.
* [API Reference](https://docs.sorsa.io/api-reference-guide): full specification for `/list-tweets`, `/user-tweets`, `/search-tweets`, and all Sorsa API endpoints.