X (Twitter) WebSocket API quick start

# pip install "websockets>=13"   (Python 3.9+)
import asyncio, json, os
from websockets.asyncio.client import connect

# Normal plans. X Ultimate's selected accounts: wss://ws.ultimate.1322.io/ws/ultimate
WS_URL = "wss://ws.normal.1322.io/ws/normal"
HEADERS = {"X-API-Key": os.environ["X_API_KEY"]}

def handle(event):
    tweet = event.get("tweet") or {}
    author = tweet.get("author") or {}
    text = (tweet.get("body") or {}).get("text") or ""
    # Stages of one tweet share tweet["id"]: merge them instead of posting twice.
    print(event.get("type"), tweet.get("id"), author.get("handle"), text[:80])

async def main():
    backoff = 1
    while True:
        try:
            async with connect(WS_URL, additional_headers=HEADERS) as ws:
                backoff = 1
                async for raw in ws:
                    try:
                        event = json.loads(raw)
                    except ValueError:
                        continue
                    if isinstance(event, dict):
                        handle(event)
        except Exception as exc:
            print("disconnected:", exc)
        await asyncio.sleep(backoff)
        backoff = min(backoff * 2, 30)

asyncio.run(main())

// npm install ws, save as consumer.mjs, run: node consumer.mjs   (Node 18+)
import WebSocket from "ws"

// Normal plans. X Ultimate's selected accounts: wss://ws.ultimate.1322.io/ws/ultimate
const WS_URL = "wss://ws.normal.1322.io/ws/normal"
const API_KEY = process.env.X_API_KEY
if (!API_KEY) throw new Error("Set X_API_KEY")

function handle(event) {
  const tweet = event.tweet ?? {}
  // Stages of one tweet share tweet.id: merge them instead of posting twice.
  console.log(event.type, tweet.id, tweet.author?.handle, (tweet.body?.text ?? "").slice(0, 80))
}

let backoff = 1000

function connect() {
  const ws = new WebSocket(WS_URL, { headers: { "X-API-Key": API_KEY } })
  ws.on("open", () => { backoff = 1000 })
  ws.on("message", (raw) => {
    let event
    try { event = JSON.parse(raw.toString()) } catch { return }
    if (event && typeof event === "object" && !Array.isArray(event)) handle(event)
  })
  ws.on("error", (err) => console.error("socket error:", err.message))
  ws.on("close", () => {
    setTimeout(connect, backoff)
    backoff = Math.min(backoff * 2, 30000)
  })
}

connect()

{
  "id": "6f1d2c3b-1322-4a5e-9b7c-0d1e2f3a4b5c",
  "type": "tweet.mini.update",
  "source": "1322",
  "tweet": {
    "id": "1978456123456789012",
    "type": "TWEET",
    "created_at": 1792073001000,
    "author": {
      "id": "1500000000000000001",
      "handle": "exampledesk",
      "name": "Example Desk",
      "verified": false,
      "metrics": {
        "following": 210,
        "followers": 48210
      }
    },
    "subtweet": null,
    "reply": null,
    "quoted": null,
    "body": {
      "text": "Q3 numbers are live on the blog.",
      "urls": [],
      "mentions": []
    },
    "media": {
      "images": [],
      "videos": [],
      "thumbnails": []
    }
  }
}

curl "https://api.1322.io/v1/tracked" \
  -H "X-API-Key: $X_API_KEY"

curl -X POST "https://api.1322.io/v1/tracked" \
  -H "Content-Type: application/json" \
  -H "X-API-Key: $X_API_KEY" \
  -d '{ "identifiers": "exampledesk,examplefounder", "type": "username" }'

curl -X DELETE "https://api.1322.io/v1/tracked" \
  -H "Content-Type: application/json" \
  -H "X-API-Key: $X_API_KEY" \
  -d '{ "identifiers": "examplefounder", "type": "username" }'

curl "https://api.1322.io/v1/keys" \
  -H "X-API-Key: $X_API_KEY"

curl "https://api.1322.io/v1/resolve/username/exampledesk" \
  -H "X-API-Key: $X_API_KEY"