---
title: "Complete WebSocket example"
canonical_url: "https://tweetstream.io/docs/examples"
last_updated: 2026-09-29
---

# Complete WebSocket example

Copy one TypeScript or Python program, run it with your API key, and watch it connect, reconnect, merge tweet updates, follow reference chains, and time each post.

Canonical URL: https://tweetstream.io/docs/examples

> Using an AI coding agent?
>
> Point it at this page's Markdown and the protocol specs, and ask it to follow these rules:
>
> 1. Key every post by `(platform, tweetId)`.
> 2. Merge `tweet/update` fields over the stored content. An update carries only the fields it changes.
> 3. When an update carries `ref`, replace the stored `ref` and follow `ref.subtweet` for the rest of the chain.
> 4. Treat each `tweet/meta` as a full snapshot and replace the stored `meta` with it.
> 5. Skip any `t` and `op` pair you don't handle and keep the socket open.
>
> - [This page as Markdown](https://tweetstream.io/docs/examples.md)
> - [AsyncAPI spec](https://tweetstream.io/docs/asyncapi.json)
> - [OpenAPI spec](https://tweetstream.io/docs/openapi.json)

## Run it

Replace `YOUR_API_KEY` with the key from your dashboard. A trial key works. Posts arrive when a monitored account posts, so add at least one account in the dashboard or set `TWEETSTREAM_ACCOUNTS`.

### TypeScript

Needs Node.js 22.18 or later, which runs TypeScript files directly. The program has no dependencies.

1. Copy the program

```typescript
// TweetStream complete example: connect, reconnect, handle every event, and measure latency.
// Node.js 22.18 or later runs this TypeScript file directly. No dependencies.
//
//   TWEETSTREAM_API_KEY=YOUR_API_KEY node tweetstream-example.ts
//
// Optional settings:
//   TWEETSTREAM_ACCOUNTS=handle1,handle2               add these accounts before connecting
//   TWEETSTREAM_WS_URL=wss://ws-iad.tweetstream.io/ws  use the US endpoint

// The messages this program reads, limited to the fields it uses. Every field of every event:
// https://tweetstream.io/docs/asyncapi.json

// Every server message is a v1 envelope. Route on t (channel) and op (operation); d is the payload.
type Envelope = { v: 1; t: string; op: string; id?: string; ts: number; d: unknown };

// Posts name their author by handle. Some account events use username instead.
type Account = { id?: string; handle?: string; username?: string; platform?: string };

type PostKind = "post" | "quote" | "reply" | "retweet";

// A post that this one quotes, replies to, or reposts. It can reference another through subtweet.
type Reference = {
  type: Exclude<PostKind, "post">;
  tweetId: string;
  author?: Account;
  text?: string;
  subtweet?: Reference;
};

// tweet/content: a new post. author.platform is "twitter" when it is missing.
type PostContent = {
  tweetId: string;
  text: string;
  createdAt: number; // Unix milliseconds
  kind: PostKind;
  author: Account;
  ref?: Reference;
};

// tweet/update: only the fields that changed. textUpdateType "completion" delivers the full text
// of a post that arrived shortened; it is not an edit.
type PostUpdate = Partial<Omit<PostContent, "createdAt">> & {
  tweetId: string;
  platform?: string;
  textUpdateType?: "completion";
};

// tweet/meta: a complete snapshot of the tokens, markets, and image text found in a post.
type PostMeta = {
  tweetId: string;
  platform?: string;
  detected?: {
    tokens?: Array<{ symbol?: string; contract?: string; priceUsd?: number }>;
    prediction?: Array<{ title?: string; priceUsd?: number }>;
  };
  ocr?: { text: string };
};

// tweet/delete, tweet/pin, and tweet/unpin.
type PostEvent = { tweetId: string; platform?: string; author?: Account };

type Coin = { symbol?: string; address?: string };

// The payload of each t/op pair this program handles.
type Payloads = {
  "control/auth_pong": Record<string, never>;
  "tweet/content": PostContent;
  "tweet/update": PostUpdate;
  "tweet/meta": PostMeta;
  "tweet/delete": PostEvent;
  "tweet/pin": PostEvent;
  "tweet/unpin": PostEvent;
  "account/follow": { actor: Account; target: Account };
  "account/unfollow": { actor: Account; target: Account };
  "account/profile_update": { actor: Account; changes: Record<string, unknown> };
  "account/affiliate_update": { organization: Account; member: Account; action: string };
  "account/callout": { caller: Account; token: Coin };
  "account/callout_update": { updateOf: string; token?: Coin };
  "account/thesis": { author: Account; coin: Coin; thesis: { text: string } };
  "account/token_verified": { coin: Coin };
  "news/article": { source?: { name?: string }; title?: string };
  "charity/added": { name: string };
  "control/twitter_handles_result": {
    results: Array<{ input: string; state: string }>;
    error: string | null;
  };
};
type Pair = keyof Payloads;
type Handler<P> = (d: P, receivedAt: number) => void;

// What this program keeps for each post.
type StoredPost = {
  content: Partial<PostUpdate> | null;
  meta: PostMeta | null;
  deleted: boolean;
  pinned: boolean;
  hasContent: boolean;
};

const API_KEY = process.env.TWEETSTREAM_API_KEY?.trim();
const DASHBOARD = "https://tweetstream.io/dashboard?tab=api";
if (!API_KEY) {
  console.error(`Set TWEETSTREAM_API_KEY. Copy your key from ${DASHBOARD}`);
  process.exit(1);
}
const WS_URL = process.env.TWEETSTREAM_WS_URL || "wss://ws-global.tweetstream.io/ws";
const API_URL = "https://api.tweetstream.io";
const ACCOUNTS = (process.env.TWEETSTREAM_ACCOUNTS ?? "")
  .split(",")
  .map((handle) => handle.trim().replace(/^@/, ""))
  .filter(Boolean);
const TWITTER_EPOCH_MS = 1288834974657n;

const posts = new Map<string, StoredPost>(); // "platform:tweetId" -> stored post
const counts = new Map<string, number>(); // "t/op" -> number of events seen
const latencies: number[] = []; // publish-to-receipt ms for the most recent 10,000 new posts
let attempt = 0;
let hinted = false;

const log = (label: string, text: string) => console.log(`${label.padEnd(9)} ${text}`);
const at = (account: Account | undefined) =>
  `@${String(account?.handle ?? account?.username ?? "unknown").replace(/^@/, "")}`;
const clip = (text: unknown, max = 80) => {
  const flat = String(text ?? "").replace(/\s+/g, " ").trim();
  return JSON.stringify(flat.length > max ? `${flat.slice(0, max - 1)}…` : flat);
};

// Posts are keyed by (platform, tweetId). Content carries the platform on its author;
// later operations carry it on d.platform.
const platformOf = (d: PostEvent) => d.platform ?? d.author?.platform ?? "twitter";
function postFor(d: PostEvent, platform = platformOf(d)): StoredPost {
  const key = `${platform}:${d.tweetId}`;
  let post = posts.get(key);
  if (!post) {
    post = { content: null, deleted: false, hasContent: false, meta: null, pinned: false };
    posts.set(key, post);
    if (posts.size > 10_000) {
      const oldest = posts.keys().next().value;
      if (oldest !== undefined) posts.delete(oldest); // bound memory
    }
  }
  return post;
}

// Updates carry only the fields that changed, so merge them over what you stored.
const merge = (base: Partial<PostUpdate> | null, patch: Partial<PostUpdate> | null) => ({
  ...base,
  ...patch,
  author: { ...base?.author, ...patch?.author },
});

function publishedAtMs(platform: string, d: PostContent): number | null {
  // X post IDs encode their publish time: (id >> 22) + the Twitter epoch.
  if (platform === "twitter" && /^\d+$/.test(d.tweetId)) {
    return Number((BigInt(d.tweetId) >> 22n) + TWITTER_EPOCH_MS);
  }
  return typeof d.createdAt === "number" ? d.createdAt : null;
}

function referenceChain(ref: Reference | undefined): string {
  const links: string[] = [];
  // A referenced post can reference another one through ref.subtweet.
  for (let r = ref, depth = 0; r && depth < 8; r = r.subtweet, depth++) {
    links.push(`${r.type ?? "ref"} ${at(r.author)} ${r.tweetId} ${clip(r.text, 40)}`);
  }
  return links.join(" -> ");
}

function describeMeta(meta: PostMeta): string {
  const price = (value: number | undefined, prefix = "") => (value == null ? "" : ` ${prefix}${value}`);
  const tokens = meta.detected?.tokens ?? [];
  const markets = meta.detected?.prediction ?? [];
  const parts: string[] = [];
  if (tokens.length) {
    parts.push(`tokens ${tokens.map((t) => `${t.symbol ?? t.contract}${price(t.priceUsd, "$")}`).join(", ")}`);
  }
  if (markets.length) {
    parts.push(`markets ${markets.map((m) => `${clip(m.title, 40)}${price(m.priceUsd)}`).join(", ")}`);
  }
  if (meta.ocr?.text) parts.push(`ocr ${clip(meta.ocr.text, 40)}`);
  return parts.join(" | ") || "no detections";
}

// Delete, pin, and unpin change the stored post's state and keep its content.
function lifecycle(d: PostEvent, label: string, change: Partial<StoredPost>) {
  Object.assign(postFor(d), change);
  log(label, `${at(d.author)} ${d.tweetId}`);
}

const handlers: { [P in Pair]: Handler<Payloads[P]> } = {
  "control/auth_pong": () => log("auth", "ok, waiting for events"),
  "tweet/content": (d, receivedAt) => {
    const platform = d.author?.platform ?? "twitter";
    const post = postFor(d, platform);
    post.content = merge(d, post.content); // an update that arrived first still wins
    let latency = "";
    const publishedAt = publishedAtMs(platform, d);
    if (!post.hasContent && publishedAt !== null) {
      latencies.push(receivedAt - publishedAt);
      if (latencies.length > 10_000) latencies.shift();
      latency = `${receivedAt - publishedAt} ms  `;
    }
    post.hasContent = true;
    const where = platform === "twitter" ? "" : ` (${platform})`;
    log("post", `${at(d.author)}${where} ${d.tweetId}  ${latency}${clip(post.content.text)}`);
    if (d.ref) log("", `chain: ${referenceChain(d.ref)}`);
  },
  "tweet/update": (d) => {
    const post = postFor(d);
    post.content = merge(post.content, d); // a ref in the update replaces the stored one
    const who = `${at(post.content.author)} ${d.tweetId}`;
    const changed = Object.keys(d).filter((k) => !["platform", "receivedAt", "tweetId"].includes(k));
    if (d.ref) log("update", `${who} chain: ${referenceChain(d.ref)}`);
    else if (d.textUpdateType === "completion") log("update", `${who} full text ${clip(d.text)}`);
    else log("update", `${who} changed ${changed.join(", ")}`);
  },
  "tweet/meta": (d) => {
    const post = postFor(d);
    post.meta = d; // each meta is a complete snapshot: replace it, never merge
    log("meta", `${at(post.content?.author)} ${d.tweetId}  ${describeMeta(d)}`);
  },
  "tweet/delete": (d) => lifecycle(d, "delete", { deleted: true }),
  "tweet/pin": (d) => lifecycle(d, "pin", { pinned: true }),
  "tweet/unpin": (d) => lifecycle(d, "unpin", { pinned: false }),
  "account/follow": (d) => log("follow", `${at(d.actor)} followed ${at(d.target)}`),
  "account/unfollow": (d) => log("unfollow", `${at(d.actor)} unfollowed ${at(d.target)}`),
  "account/profile_update": (d) =>
    log("profile", `${at(d.actor)} changed ${Object.keys(d.changes ?? {}).join(", ")}`),
  "account/affiliate_update": (d) =>
    log("affiliate", `${at(d.organization)} ${d.action} ${at(d.member)}`),
  "account/callout": (d) =>
    log("callout", `${at(d.caller)} called ${d.token?.symbol} ${d.token?.address}`),
  "account/callout_update": (d) => log("callout", `update to ${d.updateOf}: ${d.token?.symbol}`),
  "account/thesis": (d) =>
    log("thesis", `${at(d.author)} on ${d.coin?.symbol}: ${clip(d.thesis?.text, 60)}`),
  "account/token_verified": (d) => log("verified", `${d.coin?.symbol ?? d.coin?.address}`),
  "news/article": (d) => log("news", `${d.source?.name}: ${clip(d.title)}`),
  "charity/added": (d) => log("charity", d.name),
  "control/twitter_handles_result": (d) =>
    log("handles", (d.results ?? []).map((r) => `${r.input} ${r.state}`).join(", ") || String(d.error)),
};
const isPair = (pair: string): pair is Pair => Object.hasOwn(handlers, pair);

function onMessage(data: unknown, receivedAt: number) {
  let event: Partial<Envelope> | null;
  try {
    event = JSON.parse(String(data));
  } catch {
    log("skip", "frame is not JSON");
    return;
  }
  if (event?.v !== 1 || typeof event.t !== "string" || typeof event.op !== "string") return;
  const pair = `${event.t}/${event.op}`;
  counts.set(pair, (counts.get(pair) ?? 0) + 1);
  if (!isPair(pair)) return; // unknown pairs are counted, then ignored
  // The payload is JSON from the server; the types above describe what each pair carries.
  const handle = handlers[pair] as Handler<unknown>;
  try {
    handle(event.d ?? {}, receivedAt);
  } catch (error) {
    log("error", `${pair}: ${error}`); // one malformed event must not stop the program
  }
}

function printSummary() {
  const sorted = [...latencies].sort((a, b) => a - b);
  const pct = (p: number) => sorted[Math.max(0, Math.ceil(p * sorted.length) - 1)];
  const timing = sorted.length
    ? `p50 ${pct(0.5)} ms  p95 ${pct(0.95)} ms  (n=${sorted.length})`
    : "no posts yet";
  const seen = [...counts].map(([pair, n]) => `${pair} ${n}`).join(", ") || "none";
  log("summary", `latency ${timing} | events ${seen}`);
}

// POST /api/add-account answers with one row per handle.
type AddAccountResponse = {
  results?: Array<{ input: string; state: string; message?: string }>;
  error?: string;
};

async function addAccounts() {
  const response = await fetch(`${API_URL}/api/add-account`, {
    body: JSON.stringify({ accounts: ACCOUNTS }),
    headers: { Authorization: `Bearer ${API_KEY}`, "Content-Type": "application/json" },
    method: "POST",
    signal: AbortSignal.timeout(15_000),
  });
  const body = (await response.json().catch(() => ({}))) as AddAccountResponse;
  // One row per handle, even on 207 (some failed) or 400 (all failed). Each row says why.
  const rows = Array.isArray(body.results) ? body.results : [];
  for (const row of rows) {
    log("accounts", `${row.input} ${row.state}${row.message ? `: ${row.message}` : ""}`);
  }
  if (!rows.length && !response.ok) {
    log("accounts", `add failed (${response.status}): ${body.error ?? "unknown error"}`);
  }
}

// GET /api/me reports which pool the key connects to and how many connections each has open.
type ConnectionPool = { count: number; limit: number };
type MeResponse = {
  credentialScope?: "standard" | "ultra_speed";
  websocket?: ConnectionPool; // standard keys
  ultraSpeed?: { websocket?: ConnectionPool } | null; // Ultra Speed keys
};

async function explainRefusedConnection() {
  // Node's built-in WebSocket hides the HTTP status of a refused upgrade, so ask the REST API.
  const headers = { Authorization: `Bearer ${API_KEY}` };
  const response = await fetch(`${API_URL}/api/me`, {
    headers,
    signal: AbortSignal.timeout(10_000),
  }).catch(() => null);
  if (response?.status === 401) {
    console.error(`API key rejected (401). Copy a current key from ${DASHBOARD}`);
    process.exit(1);
  }
  if (response?.status === 403) {
    console.error("Streaming is not active for this key (403). Start a trial or renew your plan.");
    process.exit(1);
  }
  const me = (response?.ok ? await response.json().catch(() => null) : null) as MeResponse | null;
  const pool = me?.credentialScope === "ultra_speed" ? me.ultraSpeed?.websocket : me?.websocket;
  if (pool && pool.count >= pool.limit) {
    log("limit", `all ${pool.limit} connections are in use; close another client`);
  }
}

function connect() {
  const socket = new WebSocket(WS_URL, ["tweetstream.v1", `tweetstream.auth.token.${API_KEY}`]);
  let openedAt: number | null = null; // set while this socket is open
  socket.addEventListener("open", () => {
    openedAt = Date.now();
    log("connected", WS_URL);
    socket.send(JSON.stringify({ op: "auth_ping" })); // answered by control/auth_pong
    setTimeout(() => {
      if (openedAt !== null && !latencies.length && !hinted) {
        hinted = true;
        log("hint", "posts appear when a monitored account posts. Add accounts in the");
        log("hint", "dashboard, or run again with TWEETSTREAM_ACCOUNTS=handle1,handle2");
      }
    }, 120_000);
  });
  // Take the receipt time before any parsing so latency excludes your own processing.
  socket.addEventListener("message", (event) => onMessage(event.data, Date.now()));
  let settled = false; // some runtimes fire both error and close; reconnect only once
  const reconnect = async (reason: string) => {
    if (settled) return;
    settled = true;
    if (openedAt === null) await explainRefusedConnection();
    else if (Date.now() - openedAt >= 30_000) attempt = 0; // healthy session: reset backoff
    openedAt = null;
    const delay = Math.min(30_000, 1_000 * 2 ** attempt) + Math.floor(Math.random() * 250);
    attempt = Math.min(attempt + 1, 5);
    log("closed", `${reason}; reconnecting in ${(delay / 1000).toFixed(1)} s`);
    setTimeout(connect, delay);
  };
  // Node's built-in WebSocket reports a refused or failed handshake with error and no close.
  socket.addEventListener("error", () => openedAt === null && reconnect("connection failed"));
  socket.addEventListener("close", (event) => reconnect(`code ${event.code}`));
}

process.on("SIGINT", () => {
  printSummary();
  process.exit(0);
});
setInterval(printSummary, 60_000);
if (ACCOUNTS.length) {
  addAccounts()
    .catch((error: Error) => log("accounts", `add failed: ${error.message}`))
    .finally(connect);
} else {
  connect();
}
```

2. Save it as `tweetstream-example.ts`

3. Run it with your key

```bash
TWEETSTREAM_API_KEY=YOUR_API_KEY node tweetstream-example.ts
```

[Raw file: tweetstream-example.ts](https://tweetstream.io/docs/examples/tweetstream-example.ts)

### Python

Needs Python 3.10 or later. The run command creates a virtual environment and installs `websockets` into it.

1. Copy the program

```python
"""TweetStream complete example: connect, reconnect, handle every event, and measure latency.

Python 3.10 or later. One dependency, installed in a virtual environment:

    python3 -m venv tweetstream-venv
    tweetstream-venv/bin/pip install 'websockets>=14'
    TWEETSTREAM_API_KEY=YOUR_API_KEY tweetstream-venv/bin/python tweetstream_example.py

Optional settings:
    TWEETSTREAM_ACCOUNTS=handle1,handle2               add these accounts before connecting
    TWEETSTREAM_WS_URL=wss://ws-iad.tweetstream.io/ws  use the US endpoint
"""

import asyncio
import collections
import json
import math
import os
import random
import sys
import time
import urllib.error
import urllib.request

if sys.version_info < (3, 10):
    sys.exit(f"This example needs Python 3.10 or later; this is {sys.version.split()[0]}.")

import websockets

API_KEY = os.environ.get("TWEETSTREAM_API_KEY", "").strip()
DASHBOARD = "https://tweetstream.io/dashboard?tab=api"
if not API_KEY:
    sys.exit(f"Set TWEETSTREAM_API_KEY. Copy your key from {DASHBOARD}")
WS_URL = os.environ.get("TWEETSTREAM_WS_URL") or "wss://ws-global.tweetstream.io/ws"
API_URL = "https://api.tweetstream.io"
ACCOUNTS = [
    handle.strip().lstrip("@")
    for handle in os.environ.get("TWEETSTREAM_ACCOUNTS", "").split(",")
    if handle.strip()
]
TWITTER_EPOCH_MS = 1288834974657

posts = {}  # (platform, tweetId) -> {"content", "meta", "deleted", "pinned", "has_content"}
counts = {}  # "t/op" -> number of events seen
latencies = collections.deque(maxlen=10_000)  # publish-to-receipt ms, most recent new posts
hinted = False


def log(label, text):
    print(f"{label:<9} {text}", flush=True)


def at(account):
    account = account if isinstance(account, dict) else {}
    return "@" + str(account.get("handle") or account.get("username") or "unknown").lstrip("@")


def clip(text, limit=80):
    flat = " ".join(str(text or "").split())
    return json.dumps(flat if len(flat) <= limit else flat[: limit - 1] + "…", ensure_ascii=False)


def field(d, name, key):
    return (d.get(name) or {}).get(key)


# Posts are keyed by (platform, tweetId). Content carries the platform on its author;
# later operations carry it on d.platform.
def post_for(d, platform=None):
    platform = platform or d.get("platform") or field(d, "author", "platform") or "twitter"
    key = (platform, d["tweetId"])
    if key not in posts:
        posts[key] = {"content": None, "meta": None, "deleted": False, "pinned": False}
        posts[key]["has_content"] = False
        if len(posts) > 10_000:
            posts.pop(next(iter(posts)))  # bound memory
    return posts[key]


def merge(base, patch):
    # Updates carry only the fields that changed, so merge them over what you stored.
    base, patch = base or {}, patch or {}
    return {**base, **patch, "author": {**(base.get("author") or {}), **(patch.get("author") or {})}}


def published_at_ms(platform, d):
    tweet_id = str(d.get("tweetId", ""))
    if platform == "twitter" and tweet_id.isdigit():
        return (int(tweet_id) >> 22) + TWITTER_EPOCH_MS  # X post IDs encode their publish time
    created_at = d.get("createdAt")
    return created_at if isinstance(created_at, (int, float)) else None


def reference_chain(ref):
    links = []
    # A referenced post can reference another one through subtweet.
    while ref and len(links) < 8:
        text = clip(ref.get("text"), 40)
        links.append(f"{ref.get('type', 'ref')} {at(ref.get('author'))} {ref.get('tweetId')} {text}")
        ref = ref.get("subtweet")
    return " -> ".join(links)


def describe_meta(meta):
    detected = meta.get("detected") or {}
    tokens = [t.get("symbol") or t.get("contract") for t in detected.get("tokens") or []]
    prices = [t.get("priceUsd") for t in detected.get("tokens") or []]
    markets = [clip(m.get("title"), 40) for m in detected.get("prediction") or []]
    odds = [m.get("priceUsd") for m in detected.get("prediction") or []]
    parts = []
    if tokens:
        parts.append("tokens " + ", ".join(t if p is None else f"{t} ${p}" for t, p in zip(tokens, prices)))
    if markets:
        parts.append("markets " + ", ".join(m if p is None else f"{m} {p}" for m, p in zip(markets, odds)))
    if field(meta, "ocr", "text"):
        parts.append(f"ocr {clip(meta['ocr']['text'], 40)}")
    return " | ".join(parts) or "no detections"


def on_content(d, received_at):
    platform = field(d, "author", "platform") or "twitter"
    post = post_for(d, platform)
    post["content"] = merge(d, post["content"])  # an update that arrived first still wins
    latency = ""
    published_at = published_at_ms(platform, d)
    if not post["has_content"] and published_at is not None:
        latencies.append(round(received_at - published_at))
        latency = f"{latencies[-1]} ms  "
    post["has_content"] = True
    where = "" if platform == "twitter" else f" ({platform})"
    text = clip(post["content"].get("text"))
    log("post", f"{at(d.get('author'))}{where} {d['tweetId']}  {latency}{text}")
    if d.get("ref"):
        log("", f"chain: {reference_chain(d['ref'])}")


def on_update(d, _):
    post = post_for(d)
    post["content"] = merge(post["content"], d)  # a ref in the update replaces the stored one
    who = f"{at(post['content'].get('author'))} {d['tweetId']}"
    changed = ", ".join(k for k in d if k not in ("platform", "receivedAt", "tweetId"))
    if d.get("ref"):
        log("update", f"{who} chain: {reference_chain(d['ref'])}")
    elif d.get("textUpdateType") == "completion":
        log("update", f"{who} full text {clip(d.get('text'))}")
    else:
        log("update", f"{who} changed {changed}")


def on_meta(d, _):
    post = post_for(d)
    post["meta"] = d  # each meta is a complete snapshot: replace it, never merge
    log("meta", f"{at((post['content'] or {}).get('author'))} {d['tweetId']}  {describe_meta(d)}")


def lifecycle(label, **change):
    # Delete, pin, and unpin change the stored post's state and keep its content.
    def handle(d, _):
        post_for(d).update(change)
        log(label, f"{at(d.get('author'))} {d['tweetId']}")

    return handle


HANDLERS = {
    "tweet/content": on_content,
    "tweet/update": on_update,
    "tweet/meta": on_meta,
    "tweet/delete": lifecycle("delete", deleted=True),
    "tweet/pin": lifecycle("pin", pinned=True),
    "tweet/unpin": lifecycle("unpin", pinned=False),
}


def event_line(pair, d):
    # One readable line for account, news, charity, and control events.
    actor, target = at(d.get("actor")), at(d.get("target"))
    token, coin = d.get("token") or {}, d.get("coin") or {}
    results = ", ".join(f"{r.get('input')} {r.get('state')}" for r in d.get("results") or [])
    return {
        "control/auth_pong": ("auth", "ok, waiting for events"),
        "control/twitter_handles_result": ("handles", results or str(d.get("error"))),
        "account/follow": ("follow", f"{actor} followed {target}"),
        "account/unfollow": ("unfollow", f"{actor} unfollowed {target}"),
        "account/profile_update": ("profile", f"{actor} changed {', '.join(d.get('changes') or {})}"),
        "account/affiliate_update": (
            "affiliate",
            f"{at(d.get('organization'))} {d.get('action')} {at(d.get('member'))}",
        ),
        "account/callout": (
            "callout",
            f"{at(d.get('caller'))} called {token.get('symbol')} {token.get('address')}",
        ),
        "account/callout_update": ("callout", f"update to {d.get('updateOf')}: {token.get('symbol')}"),
        "account/thesis": (
            "thesis",
            f"{at(d.get('author'))} on {coin.get('symbol')}: {clip(field(d, 'thesis', 'text'), 60)}",
        ),
        "account/token_verified": ("verified", coin.get("symbol") or coin.get("address")),
        "news/article": ("news", f"{field(d, 'source', 'name')}: {clip(d.get('title'))}"),
        "charity/added": ("charity", d.get("name")),
    }.get(pair)


def on_message(raw, received_at):
    try:
        event = json.loads(raw)
    except ValueError:
        return log("skip", "frame is not JSON")
    # Envelope: {v, t, op, id?, ts, d}. Route by t, then op.
    if not isinstance(event, dict) or event.get("v") != 1:
        return
    pair = f"{event.get('t')}/{event.get('op')}"
    counts[pair] = counts.get(pair, 0) + 1
    d = event.get("d") or {}
    try:
        if pair in HANDLERS:
            HANDLERS[pair](d, received_at)
        elif event_line(pair, d):  # unknown pairs are counted, then ignored
            log(*event_line(pair, d))
    except Exception as error:  # one malformed event must not close the connection
        log("error", f"{pair}: {error!r}")


def print_summary():
    ordered = sorted(latencies)
    pct = lambda p: ordered[max(0, math.ceil(p * len(ordered)) - 1)]
    timing = f"p50 {pct(0.5)} ms  p95 {pct(0.95)} ms  (n={len(ordered)})" if ordered else "no posts yet"
    seen = ", ".join(f"{pair} {n}" for pair, n in counts.items()) or "none"
    log("summary", f"latency {timing} | events {seen}")


def add_accounts():
    request = urllib.request.Request(
        f"{API_URL}/api/add-account",
        data=json.dumps({"accounts": ACCOUNTS}).encode(),
        headers={"Authorization": f"Bearer {API_KEY}", "Content-Type": "application/json"},
        method="POST",
    )
    try:
        with urllib.request.urlopen(request, timeout=15) as response:
            status, raw = response.status, response.read()
    except urllib.error.HTTPError as error:  # 400, 429, and 503 can still carry results
        status, raw = error.code, error.read()
    except OSError as error:
        return log("accounts", f"add failed: {error}")
    try:
        body = json.loads(raw)
    except ValueError:
        body = {}
    body = body if isinstance(body, dict) else {}
    # One row per handle, even on 207 (some failed) or 400 (all failed). Each row says why.
    rows = body.get("results") if isinstance(body.get("results"), list) else []
    for row in rows:
        message = f": {row['message']}" if row.get("message") else ""
        log("accounts", f"{row.get('input')} {row.get('state')}{message}")
    if not rows and status >= 300:
        log("accounts", f"add failed ({status}): {body.get('error') or 'unknown error'}")


async def report_every_minute():
    while True:
        await asyncio.sleep(60)
        print_summary()


async def hint_when_quiet():
    global hinted
    await asyncio.sleep(120)
    if not latencies and not hinted:
        hinted = True
        log("hint", "posts appear when a monitored account posts. Add accounts in the")
        log("hint", "dashboard, or run again with TWEETSTREAM_ACCOUNTS=handle1,handle2")


async def main():
    asyncio.create_task(report_every_minute())
    protocols = ["tweetstream.v1", f"tweetstream.auth.token.{API_KEY}"]
    attempt = 0
    while True:
        opened_at, reason = None, "connection closed"
        try:
            async with websockets.connect(WS_URL, subprotocols=protocols) as ws:
                opened_at = time.monotonic()
                log("connected", WS_URL)
                await ws.send(json.dumps({"op": "auth_ping"}))  # answered by control/auth_pong
                hint = asyncio.create_task(hint_when_quiet())
                try:
                    async for raw in ws:
                        # Take the receipt time before parsing so latency excludes your own work.
                        on_message(raw, time.time() * 1000)
                finally:
                    hint.cancel()
        except Exception as error:
            # A refused upgrade carries its HTTP status on error.response. 429 means the pool
            # this key connects to (standard or Ultra Speed) has no free connection.
            status = getattr(getattr(error, "response", None), "status_code", None)
            if status == 401:
                sys.exit(f"API key rejected (401). Copy a current key from {DASHBOARD}")
            if status == 403:
                sys.exit("Streaming is not active for this key (403). Start a trial or renew your plan.")
            reason = "all connections are in use" if status == 429 else f"{type(error).__name__}: {error}"
        if opened_at is not None and time.monotonic() - opened_at >= 30:
            attempt = 0  # a healthy session resets backoff
        delay = min(30, 2**attempt) + random.random() / 4
        attempt = min(attempt + 1, 5)
        log("closed", f"{reason}; reconnecting in {delay:.1f} s")
        await asyncio.sleep(delay)


if __name__ == "__main__":
    if ACCOUNTS:
        add_accounts()
    try:
        asyncio.run(main())
    except KeyboardInterrupt:
        print_summary()
```

2. Save it as `tweetstream_example.py`

3. Run it with your key

```bash
python3 -m venv tweetstream-venv && \
tweetstream-venv/bin/pip install 'websockets>=14' && \
TWEETSTREAM_API_KEY=YOUR_API_KEY tweetstream-venv/bin/python tweetstream_example.py
```

[Raw file: tweetstream_example.py](https://tweetstream.io/docs/examples/tweetstream_example.py)

### Expected output

```text
connected wss://ws-global.tweetstream.io/ws
auth      ok, waiting for events
post      @marketdesk 2104934217937850514  183 ms  "SOL breaks out of the 4h range. Watching $98.5 as…"
update    @marketdesk 2104934217937850514 chain: quote @projectteam 2104932246614966349 "Mainnet is live." -> reply @devteam 2104929205744566281 "Ship date?"
update    @marketdesk 2104934217937850514 full text "SOL breaks out of the 4h range. Watching $98.5 as support. Chart below."
meta      @marketdesk 2104934217937850514  tokens SOL $98.5 | ocr "SOL/USD 4h"
delete    @marketdesk 2104934217937850514
follow    @marketdesk followed @projectteam
summary   latency p50 183 ms  p95 183 ms  (n=1) | events control/auth_pong 1, tweet/content 1, tweet/update 2, tweet/meta 1, tweet/delete 1, account/follow 1
```

## The event sequence

These frames produce the expected output above. Their timestamps are illustrative. Each post starts with `tweet/content`. A `tweet/update` for the same `(platform, tweetId)` adds fields to it, and its `ref` links the post to the posts it quotes or replies to.

1. You send `auth_ping`. Sent as soon as the socket opens, to check the key.

```json
{
  "op": "auth_ping"
}
```

2. Server sends `control/auth_pong`. The key works. The program prints an `auth` line and waits for events.

```json
{
  "v": 1,
  "t": "control",
  "op": "auth_pong",
  "ts": 1790690396000,
  "d": {}
}
```

3. Server sends `tweet/content`. A new post, here with shortened text. The program stores it under `(platform, tweetId)` and prints it with its latency.

```json
{
  "v": 1,
  "t": "tweet",
  "op": "content",
  "id": "2104934217937850514",
  "ts": 1790690400160,
  "d": {
    "tweetId": "2104934217937850514",
    "text": "SOL breaks out of the 4h range. Watching $98.5 as…",
    "createdAt": 1790690400000,
    "receivedAt": 1790690400150,
    "kind": "quote",
    "author": {
      "id": "1450012345678901248",
      "handle": "@marketdesk",
      "name": "Market Desk",
      "platform": "twitter"
    }
  }
}
```

4. Server sends `tweet/update`. Adds a reference to the same post. The program merges the fields over the stored post, replaces its `ref`, and prints the chain by following `ref.subtweet`.

```json
{
  "v": 1,
  "t": "tweet",
  "op": "update",
  "id": "2104934217937850514",
  "ts": 1790690402000,
  "d": {
    "tweetId": "2104934217937850514",
    "platform": "twitter",
    "receivedAt": 1790690401990,
    "kind": "quote",
    "ref": {
      "type": "quote",
      "tweetId": "2104932246614966349",
      "text": "Mainnet is live.",
      "author": {
        "id": "1380123456789012480",
        "handle": "@projectteam",
        "name": "Project Team"
      },
      "subtweet": {
        "type": "reply",
        "tweetId": "2104929205744566281",
        "text": "Ship date?",
        "author": {
          "id": "1290123456789012481",
          "handle": "@devteam",
          "name": "Dev Team"
        }
      }
    }
  }
}
```

5. Server sends `tweet/update`. Delivers the full text, with `textUpdateType` set to `completion`. The program merges it over the stored post and keeps the `ref` it already has.

```json
{
  "v": 1,
  "t": "tweet",
  "op": "update",
  "id": "2104934217937850514",
  "ts": 1790690403000,
  "d": {
    "tweetId": "2104934217937850514",
    "platform": "twitter",
    "receivedAt": 1790690402990,
    "text": "SOL breaks out of the 4h range. Watching $98.5 as support. Chart below.",
    "textUpdateType": "completion"
  }
}
```

6. Server sends `tweet/meta`. Enrichment for the post. The program replaces the stored `meta` with this snapshot and prints tokens, markets, and OCR text.

```json
{
  "v": 1,
  "t": "tweet",
  "op": "meta",
  "id": "2104934217937850514",
  "ts": 1790690404000,
  "d": {
    "tweetId": "2104934217937850514",
    "platform": "twitter",
    "receivedAt": 1790690403990,
    "detected": {
      "tokens": [
        {
          "chain": "solana",
          "contract": "So11111111111111111111111111111111111111112",
          "name": "Wrapped SOL",
          "networkId": 1399811149,
          "priceUsd": 98.5,
          "sources": [
            "text",
            "ocr"
          ],
          "symbol": "SOL"
        }
      ]
    },
    "ocr": {
      "text": "SOL/USD 4h"
    }
  }
}
```

7. Server sends `tweet/delete`. The author deleted the post. The program marks it deleted and keeps its content.

```json
{
  "v": 1,
  "t": "tweet",
  "op": "delete",
  "id": "2104934217937850514",
  "ts": 1790690430000,
  "d": {
    "eventId": "delete.264da58762fac7204da8",
    "tweetId": "2104934217937850514",
    "platform": "twitter",
    "author": {
      "id": "1450012345678901248",
      "handle": "@marketdesk"
    },
    "deletedAt": 1790690429990,
    "receivedAt": 1790690429990
  }
}
```

8. Server sends `account/follow`. A monitored account followed someone. The program prints one line and leaves stored posts alone.

```json
{
  "v": 1,
  "t": "account",
  "op": "follow",
  "id": "follow-1450012345678901248-1380123456789012480",
  "ts": 1790690438000,
  "d": {
    "actor": {
      "id": "1450012345678901248",
      "handle": "@marketdesk"
    },
    "target": {
      "id": "1380123456789012480",
      "handle": "@projectteam"
    },
    "kind": "FOLLOW",
    "eventId": "follow-1450012345678901248-1380123456789012480",
    "observedAt": 1790690437990,
    "receivedAt": 1790690437990
  }
}
```

## Connect and reconnect

`connect()` opens the socket with two subprotocols, `tweetstream.v1` and `tweetstream.auth.token.<API_KEY>`, then sends `auth_ping`. The `control/auth_pong` reply confirms the key.

After a disconnect the program waits 1, 2, 4, 8, 16, and then 30 seconds between attempts, plus a little jitter. A session that stays up for 30 seconds resets the delay.

A `401` or `403` stops the program with a message, because retrying will not fix the key or the plan. Node.js does not expose the HTTP status of a refused connection, so the TypeScript program asks `GET /api/me` for the reason. The Python program reads the status from the `websockets` error.

## Add accounts

When `TWEETSTREAM_ACCOUNTS` has handles, `addAccounts()` sends them to `POST https://api.tweetstream.io/api/add-account` before connecting. A `207` response means some handles were added and some were not, so the program prints one line per handle with its `state`.

Added accounts stay on your watchlist after the program exits.

## Posts and update chains

The program stores each post by `(platform, tweetId)`. `tweet/content` creates the post, and `tweet/update` merges in only the fields it carries, so the text and author you already have stay in place.

When an update includes `ref`, the program replaces the stored reference and prints the chain by following `ref.subtweet` from one referenced post to the next. An update can arrive before its content; the merge keeps the update's fields. `tweet/delete`, `tweet/pin`, and `tweet/unpin` change the stored post's state and leave its content alone.

### Merge updates and print the chain (TypeScript)

```typescript
// Updates carry only the fields that changed, so merge them over what you stored.
const merge = (base: Partial<PostUpdate> | null, patch: Partial<PostUpdate> | null) => ({
  ...base,
  ...patch,
  author: { ...base?.author, ...patch?.author },
});

function referenceChain(ref: Reference | undefined): string {
  const links: string[] = [];
  // A referenced post can reference another one through ref.subtweet.
  for (let r = ref, depth = 0; r && depth < 8; r = r.subtweet, depth++) {
    links.push(`${r.type ?? "ref"} ${at(r.author)} ${r.tweetId} ${clip(r.text, 40)}`);
  }
  return links.join(" -> ");
}

  "tweet/update": (d) => {
    const post = postFor(d);
    post.content = merge(post.content, d); // a ref in the update replaces the stored one
    const who = `${at(post.content.author)} ${d.tweetId}`;
    const changed = Object.keys(d).filter((k) => !["platform", "receivedAt", "tweetId"].includes(k));
    if (d.ref) log("update", `${who} chain: ${referenceChain(d.ref)}`);
    else if (d.textUpdateType === "completion") log("update", `${who} full text ${clip(d.text)}`);
    else log("update", `${who} changed ${changed.join(", ")}`);
  },
```

### Merge updates and print the chain (Python)

```python
def merge(base, patch):
    # Updates carry only the fields that changed, so merge them over what you stored.
    base, patch = base or {}, patch or {}
    return {**base, **patch, "author": {**(base.get("author") or {}), **(patch.get("author") or {})}}


def reference_chain(ref):
    links = []
    # A referenced post can reference another one through subtweet.
    while ref and len(links) < 8:
        text = clip(ref.get("text"), 40)
        links.append(f"{ref.get('type', 'ref')} {at(ref.get('author'))} {ref.get('tweetId')} {text}")
        ref = ref.get("subtweet")
    return " -> ".join(links)


def on_update(d, _):
    post = post_for(d)
    post["content"] = merge(post["content"], d)  # a ref in the update replaces the stored one
    who = f"{at(post['content'].get('author'))} {d['tweetId']}"
    changed = ", ".join(k for k in d if k not in ("platform", "receivedAt", "tweetId"))
    if d.get("ref"):
        log("update", f"{who} chain: {reference_chain(d['ref'])}")
    elif d.get("textUpdateType") == "completion":
        log("update", f"{who} full text {clip(d.get('text'))}")
    else:
        log("update", f"{who} changed {changed}")
```

## Enrichment

`tweet/meta` can arrive after the post. Each one is a complete snapshot, so the program replaces the stored metadata instead of merging it.

The `meta` line lists detected tokens with prices, prediction markets with prices, and OCR text read from images, when present.

## Other events

Follows, unfollows, profile changes, affiliate changes, callouts, FOMO theses and token verifications, news articles, and charity notices each print one line. Callouts, news, charity, and FOMO events arrive only after you turn them on in the dashboard.

An unknown `t` and `op` pair is counted and skipped, and the connection stays open. If a handler fails, the program prints an `error` line and keeps reading.

## Measure latency

The message handler reads the clock before it parses anything. For X posts, the publish time comes from the post ID: shift the ID right by 22 bits and add the Twitter epoch, `1288834974657`. Other platforms use `createdAt`.

The difference is the time from publication to receipt on your machine, including the network path to you. A replayed post does not add a second sample. Every 60 seconds and when you press Ctrl+C, the program prints p50 and p95 over the most recent 10,000 posts, plus a count for each event type.

### Read the publish time (TypeScript)

```typescript
function publishedAtMs(platform: string, d: PostContent): number | null {
  // X post IDs encode their publish time: (id >> 22) + the Twitter epoch.
  if (platform === "twitter" && /^\d+$/.test(d.tweetId)) {
    return Number((BigInt(d.tweetId) >> 22n) + TWITTER_EPOCH_MS);
  }
  return typeof d.createdAt === "number" ? d.createdAt : null;
}
```

### Read the publish time (Python)

```python
def published_at_ms(platform, d):
    tweet_id = str(d.get("tweetId", ""))
    if platform == "twitter" and tweet_id.isdigit():
        return (int(tweet_id) >> 22) + TWITTER_EPOCH_MS  # X post IDs encode their publish time
    created_at = d.get("createdAt")
    return created_at if isinstance(created_at, (int, float)) else None
```

## Optional settings

Set these environment variables next to `TWEETSTREAM_API_KEY` in the run command.

| Variable | Default | Use |
| --- | --- | --- |
| `TWEETSTREAM_API_KEY` | Required | Your API key. The program never prints it. |
| `TWEETSTREAM_ACCOUNTS` | None | Comma-separated handles to add before connecting, such as `marketdesk,projectteam` |
| `TWEETSTREAM_WS_URL` | `wss://ws-global.tweetstream.io/ws` | Use `wss://ws-iad.tweetstream.io/ws` when you run in the USA |

## Common mistakes

- Treating `tweet/update` as a new post. It belongs to the post you stored under the same `(platform, tweetId)`.
- Replacing the stored post with an update. An update carries only the fields it changes, so merge it.
- Dropping an update that arrives before its content. Keep it, and when the content arrives, merge it in without overwriting the update's fields.
- Merging a `tweet/meta` into older metadata. Each one is a full snapshot, so replace the stored one.
- Ordering or deduplicating posts by the envelope `ts` or `id`. Use `(platform, tweetId)` for identity.
- Closing the socket on an unknown `t` and `op` pair. Skip the frame and keep reading.
- Retrying a `401` or `403` in a loop. A new connection will not fix the key or the plan.

## Troubleshooting

- You only see `auth` and `summary` lines. Posts arrive when a monitored account posts. Add an active account, or check your watchlist with `GET /api/me`. After two quiet minutes the program prints a hint.
- The program stops with `API key rejected (401)`. Copy the key again from the dashboard. A rotated key stops working.
- The program stops with `Streaming is not active for this key (403)`. Start a trial or renew your plan, then run it again.
- The program says all connections are in use. Your plan limits open WebSocket connections. Stop another client, or wait for an old connection to close.
- Latency is negative or much higher than expected. The number compares your clock with the post's publish time, so sync your clock with NTP. Measure from the machine that runs your trading system.
- Node.js reports `Unknown file extension ".ts"`. Your Node.js is older than 22.18. Update Node.js and run the command again.
- Python reports `No module named 'websockets'`. Start the program with `tweetstream-venv/bin/python` as the run command shows. That environment has `websockets` installed.
- Creating the virtual environment fails with `ensurepip is not available`. On Debian or Ubuntu, install the venv module with `sudo apt install python3-venv`, then run the Python command again.
- pip reports `No matching distribution found for websockets>=14`. Your `python3` is older than 3.10. Install Python 3.10 or later and create the virtual environment with it.

## Build on it

When you turn this program into a long-running consumer, start with the typed router and the reference consumer.

- [Route every operation](https://tweetstream.io/docs/production-consumer#complete-router): Typed handlers for all 18 server operations.
- [Build the reference consumer](https://tweetstream.io/docs/production-consumer#reference-consumer): Reconnects, durable recovery, and account management, one page each.
- [Open-source Solana sniper bot](https://github.com/tweetstream-io/solana-sniper-bot): A TypeScript and Rust bot built on this feed that buys on posts, bio changes, follows, and callouts.