Real-Time Streaming

Subscribe to live tweet, follower, and mention events via polling + Socket.IO, or push events straight off x.com's own event pipeline. No API keys needed.

Overview

The streaming system provides near-real-time monitoring of Twitter/X activity by polling at configurable intervals and emitting events through Socket.IO. It uses:

• Bull queue (Redis) for reliable scheduled polling
• Browser pool for managed Puppeteer instances
• Redis for deduplication and state persistence
• Socket.IO for real-time event emission to clients

Quick Start

Node.js

import { createStream, stopStream, listStreams, setIO } from 'xactions/streaming';

// Optional: connect Socket.IO for real-time events
import { Server } from 'socket.io';
const io = new Server(httpServer);
setIO(io);

// Start a tweet stream
const stream = await createStream({
  type: 'tweet',
  username: 'elonmusk',
  interval: 60,         // Poll every 60 seconds (default)
  authToken: 'your_auth_token'
});

console.log(stream);
// { id: 'abc123', type: 'tweet', username: 'elonmusk', status: 'active', interval: 60000 }

// List active streams
const streams = await listStreams();

// Stop a stream
await stopStream(stream.id);

MCP (AI Agents)

"Start monitoring @elonmusk's tweets in real-time"
→ Uses x_stream_start tool

"Show me all active streams"
→ Uses x_stream_list tool

"Stop monitoring @elonmusk"
→ Uses x_stream_stop tool

API

# Start a stream
curl -X POST http://localhost:3001/api/streams/start \
  -H "Content-Type: application/json" \
  -d '{"type": "tweet", "username": "elonmusk", "interval": 60}'

# List streams
curl http://localhost:3001/api/streams

# Stop a stream
curl -X POST http://localhost:3001/api/streams/stop \
  -d '{"streamId": "abc123"}'

Stream Types

A stream can also run on the live transport instead of polling, which adds the stream:engagement, stream:dm and stream:typing events. See Live event stream.

Live event stream

Polling asks x.com "anything new?" every 15 seconds or more. The live pipeline is the other direction: x.com's own web client keeps one long-lived connection open to https://api.x.com/live_pipeline/events and receives frames as things happen. XActions speaks the same protocol, so engagement counters, DM updates and typing indicators arrive in under a second with no polling interval at all.

Two things to know before you build on it:

• It needs a logged-in session. Guest tokens are rejected. The connection carries the auth_token and ct0 cookies, exactly like every other authed call in the toolkit. Without them open() fails with LivePipelineAuthError before a request is made.
• It is not a WebSocket, whatever its reputation says. The endpoint answers an Upgrade: websocket request the same way it answers a plain GET, and the web client reads it as a chunked HTTP response of newline-delimited JSON. createLivePipeline() speaks that transport. Everything a WebSocket would give you (push delivery, live subscription changes, automatic reconnect) is here; the wire format underneath is just HTTP.

Quick start

import { TwitterHttpClient } from 'xactions/scrapers/twitter/http';
import { createLivePipeline, Topic } from 'xactions/streaming';

// auth_token and ct0 from a logged-in session
const client = new TwitterHttpClient({ cookies: process.env.X_COOKIES });

const pipeline = createLivePipeline({
  client,
  topics: [Topic.tweetEngagement('1749528513')],
  onEvent: (event) => {
    if (event.type === 'engagement') {
      console.log(`likes ${event.payload.likeCount}, views ${event.payload.viewCount}`);
    }
  },
  onError: (err, info) => {
    console.error(err.message, info); // { fatal, willRetry, attempt, delayMs }
  },
});

const { sessionId } = await pipeline.open();
console.log('live pipeline session', sessionId);

// Change what you are watching without dropping the connection
await pipeline.subscribe(Topic.tweetEngagement('1765829534'));
await pipeline.unsubscribe(Topic.tweetEngagement('1749528513'));

// Resolves once the connection is shut and every timer is cleared
await pipeline.close();

Topics

A topic names one thing to watch. Build them with the Topic helpers rather than by hand, so a missing id fails at the call site instead of silently subscribing to nothing.

A conversation id is either a group id (1234567890) or partnerId-yourId (1234567890-9876543210).

DM topics only attach on the opening connection. x.com answers a mid-session subscribe() for a dm_update or dm_typing topic with an entry in the subscription error list, so pass them to createLivePipeline({ topics }) rather than adding them later. Engagement topics can be added and removed at any time.

Event shape

Every frame is normalised into the same object, and the untouched frame is kept on raw so nothing x.com sends is lost:

{
  type: 'engagement',                       // engagement | dm | typing | config | unknown
  topic: '/tweet_engagement/1749528513',
  payload: { tweetId: '1749528513', likeCount: 12, retweetCount: 3,
             quoteCount: 1, replyCount: 0, viewCount: 4096,
             viewCountState: 'EnabledWithCount' },
  receivedAt: '2026-08-27T20:11:04.512Z',
  raw: { topic: '...', payload: { tweet_engagement: { like_count: '12', ... } } }
}

One frame can carry several payload keys, and each one becomes its own event.

Resilience

Using it from the stream manager

createStream() takes a transport option. It defaults to poll, so existing streams behave exactly as before.

import { createStream } from 'xactions/streaming';

const stream = await createStream({
  type: 'tweet',
  username: 'elonmusk',
  transport: 'live',
  topics: ['/tweet_engagement/1749528513'],
  cookies: process.env.X_COOKIES,   // auth_token AND ct0
});

Events reach Socket.IO clients and the stream history under stream:engagement, stream:dm and stream:typing, alongside the polling events, each tagged transport: 'live':

socket.on('stream:engagement', (data) => {
  // { streamId, username, transport: 'live', topic, data: { likeCount, ... }, timestamp }
});

Polling is always the fallback. If the pipeline cannot open (no topics, no logged-in session, x.com unreachable) or gives up reconnecting later, the manager logs the reason once, sets transportFallbackReason on the stream, and schedules the normal poll job. A live stream never leaves you with no data.

updateStream(streamId, { topics }) diffs the topic set and applies it to the running session, so a watch list can change without a reconnect.

API reference

What x.com may change

This is x.com's internal pipeline, not a documented API. The endpoint paths, the frame keys, and the topic shapes here were read from the live service and can change without notice. Two design choices keep that from being a cliff: an unrecognised payload key arrives as an unknown event with the raw frame attached rather than being dropped, and the stream manager falls back to polling whenever the pipeline stops working. If frames stop arriving, check pipeline.stats.malformedFrames and the raw payload of an unknown event before assuming the session is at fault.

Configuration

Environment Variables

Interval Limits

Socket.IO Events

Clients can subscribe to real-time events:

// Client-side
const socket = io('http://localhost:3001');

socket.on('stream:tweet:new', (data) => {
  console.log('New tweet:', data);
  // { streamId, username, tweet: { text, author, timestamp, url } }
});

socket.on('stream:follower:new', (data) => {
  console.log('New follower:', data);
  // { streamId, username, follower: { username, name } }
});

socket.on('stream:follower:lost', (data) => {
  console.log('Lost follower:', data);
  // { streamId, username, unfollower: { username } }
});

socket.on('stream:mention:new', (data) => {
  console.log('New mention:', data);
  // { streamId, username, mention: { text, author, timestamp } }
});

socket.on('stream:error', (data) => {
  console.log('Stream error:', data);
  // { streamId, error, consecutiveErrors }
});

API Reference

Stream Management

Browser Pool

Poll Functions

Constants

Redis Data Model

Streams persist state in Redis with a 7-day TTL:

xactions:stream:{streamId}:state    → { status, lastPoll, consecutiveErrors, ... }
xactions:stream:{streamId}:history  → List of recent events (capped)
xactions:stream:{streamId}:meta     → { type, username, createdAt, ... }
xactions:stream:{streamId}:lock     → Distributed lock for poll coordination

Architecture

┌─────────────┐     ┌──────────────┐     ┌─────────────┐
│  Client App  │◄───│  Socket.IO   │◄───│ Stream Mgr  │
└─────────────┘     └──────────────┘     └──────┬──────┘
                                                │
                    ┌──────────────┐     ┌──────▼──────┐
                    │ Browser Pool │◄───│  Bull Queue  │
                    │ (Puppeteer)  │     │   (Redis)    │
                    └──────┬───────┘     └─────────────┘
                           │
                    ┌──────▼──────┐
                    │   x.com     │
                    │  (polling)  │
                    └─────────────┘