Build Real-Time Twitter/X Tools with the Xanguard API

You are five minutes away from a working real-time Twitter/X pipeline. This tutorial walks through six concrete things you can build with the Xanguard REST + WebSocket API today — every code example below is copy-pasteable and tested. By the end of this post you will have a Telegram bot that fires on every tracked tweet, a contract-address sniper bridge, a signed webhook receiver, and a client-side convergence detector.

This is the developer-facing companion to our real-time account tracking guide. If you have not yet decided whether real-time Twitter monitoring is worth the cost, start there.

What you get

Two transports on the B2B API key, plus webhooks on consumer plans:

• REST at https://api.xanguard.tech/v1/* — manage tracked accounts, settings and subscription state.
• WebSocket at wss://api.xanguard.tech/v1/dt/realtime/ws — receive tweet alerts as they happen. Median tweet detection is about 268 ms, with p95 at 657 ms (published live). Full reconnecting clients: Python WebSocket tutorial and Node.js WebSocket tutorial.
• Webhooks: HMAC-signed HTTPS POSTs to an endpoint you control, available on paid Tweet Alerts plans in @Xanguard_bot (see Recipe 4). The B2B feed is delivered over WebSocket and REST.

Every B2B tier includes the WebSocket and REST. Tiers differ by tracked-account count and by modules: Starter streams tweets and deleted tweets, Pro adds profile changes and new followers, Enterprise adds follow/unfollow events.

Step 0: get an API key

API keys are issued through the Xanguard B2B Telegram bot. This avoids the OAuth dance most APIs force you through and ties keys to your Telegram identity for billing.

1. Open @B2B_Xanguard_bot on Telegram.
2. Send /start and pick a plan from the menu (Starter tier is enough for everything in this tutorial; the Free consumer tier on @Xanguard_bot does not include API access).
3. Your key is shown after payment; /apikey regenerates it. It looks like dt_abcdef0123... and is shown to you exactly once — store it.

From here on, $XG_KEY is your API key.

Recipe 1 — Hello, real-time

The shortest possible client: connect to the WebSocket and print every tweet from your tracked accounts. We will add accounts via the REST API first.

Add some accounts to track:

for h in elonmusk VitalikButerin cz_binance; do
  curl -X POST https://api.xanguard.tech/v1/dt/targets \
    -H "Authorization: Bearer $XG_KEY" \
    -H "Content-Type: application/json" \
    -d "{\"handle\":\"$h\"}"
done

Response:

{
  "ok": true,
  "data": {
    "handle": "elonmusk",
    "twitter_user_id": "44196397"
  }
}

Stream tweets:

import asyncio, json, os, websockets

XG_KEY = os.environ["XG_KEY"]

async def main():
    url = "wss://api.xanguard.tech/v1/dt/realtime/ws"
    async with websockets.connect(url) as ws:
        await ws.recv()  # HELLO (op 10)
        await ws.send(json.dumps({"op": 2, "d": XG_KEY}))  # LOGIN (op 2) -> READY (op 4)
        async for raw in ws:
            msg = json.loads(raw)
            if msg.get("op") != 0:  # EVENT frames only
                continue
            d = msg["d"]
            if d.get("event") != "twitter.post.new":
                continue
            t = d["data"]
            handle = d['task_info']['handle']
            print(f"[@{handle}] {t['text'][:140]}")
            print(f"  https://x.com/{handle}/status/{t['id']}\n")  # the frame has no url field

asyncio.run(main())

Run it. The next time Elon, Vitalik, or CZ posts, you will see the tweet within a fraction of a second (median detection about 270 ms).

Recipe 2 — Filter by keyword and contract address

Most use cases do not want every tweet from every tracked account. Two filters are easy to apply in your consumer:

1. Per-account keywords — only deliver tweets containing at least one keyword from the list.
2. Contracts-only mode — only deliver tweets that contain a contract address (Solana base58 or EVM hex). This is a common filter for trading bots.

import re

KEYWORDS = {"ethereum", "l2"}
SOL_RE = re.compile(r"\b[1-9A-HJ-NP-Za-km-z]{32,44}\b")
EVM_RE = re.compile(r"\b0x[a-fA-F0-9]{40}\b")

def wanted(handle: str, text: str) -> bool:
    low = text.lower()
    if handle.lower() == "vitalikbuterin":
        return any(k in low for k in KEYWORDS)   # keyword filter for one account
    return bool(SOL_RE.search(text) or EVM_RE.search(text))  # contracts-only for the rest

# inside the EVENT loop from Recipe 1:
#     if not wanted(d["task_info"]["handle"], t["text"]): continue

From here, your consumer only acts on Vitalik tweets that mention your keywords and on contract-address tweets from everyone else. A few lines of code, and you have a KOL contract-call firehose.

Recipe 3 — A complete Solana sniper bridge

This is a common first build: tweet detected → extract contract address → fire a buy through your favorite trading bot or RPC. Below is a working sniper bridge in 40 lines.

import asyncio, json, os, re, websockets
import httpx

XG_KEY = os.environ["XG_KEY"]
TRADER_WEBHOOK = os.environ["TRADER_WEBHOOK"]  # your own trading bot's HTTP endpoint

# Solana mint addresses: base58, 32-44 chars, exclude obvious false positives
SOL_RE = re.compile(r"\b([1-9A-HJ-NP-Za-km-z]{32,44})\b")
BLACKLIST = {"So11111111111111111111111111111111111111112"}  # wrapped SOL

def extract_solana_cas(text: str) -> list[str]:
    out = []
    for m in SOL_RE.finditer(text):
        ca = m.group(1)
        if ca in BLACKLIST: continue
        if not (32 <= len(ca) <= 44): continue
        out.append(ca)
    return out

async def execute_buy(ca: str, handle: str, tweet_url: str):
    async with httpx.AsyncClient(timeout=5.0) as http:
        await http.post(TRADER_WEBHOOK, json={
            "action": "buy",
            "mint": ca,
            "amount_sol": 0.1,
            "source": f"@{handle}",
            "tweet_url": tweet_url,
        })

async def main():
    url = "wss://api.xanguard.tech/v1/dt/realtime/ws"
    async with websockets.connect(url) as ws:
        await ws.recv()  # HELLO (op 10)
        await ws.send(json.dumps({"op": 2, "d": XG_KEY}))  # LOGIN (op 2) -> READY (op 4)
        async for raw in ws:
            msg = json.loads(raw)
            if msg.get("op") != 0: continue  # EVENT frames only
            d = msg["d"]
            if d.get("event") != "twitter.post.new": continue
            t = d["data"]
            handle = d["task_info"]["handle"]
            cas = extract_solana_cas(t["text"])
            if not cas: continue
            for ca in cas:
                print(f"[@{handle}] CA: {ca}")
                await execute_buy(ca, handle, f"https://x.com/{handle}/status/{t['id']}")

asyncio.run(main())

If your trading bot does not have a webhook endpoint, swap execute_buy() for a direct Jupiter swap call. The point is: from recv() on the WebSocket to post() on your trader, you add roughly 5 ms of code. The rest of the latency is detection itself, which is published live.

Recipe 4 — Webhook receiver (Tweet Alerts plans)

WebSockets are best for low-latency clients that stay connected. Webhooks suit serverless or load-balanced deployments. Webhooks come with paid Tweet Alerts plans in @Xanguard_bot, not with the B2B key: Starter includes 1 webhook, Growth 2, Pro 5, Business and Scale 10, Enterprise 25 and Ultra 50. Generate a key with /apikey in @Xanguard_bot. Every delivery is signed with HMAC-SHA256 so you can verify it.

Register the webhook:

curl -X POST https://api.xanguard.tech/v1/webhooks \
  -H "Authorization: Bearer $XG_TWEET_ALERTS_KEY" \
  -H "Content-Type: application/json" \
  -d '{"url": "https://yourdomain.com/xg/tweet"}'

The response includes a secret — you only see it once, store it.

Receiver (FastAPI):

import hmac, hashlib, os
from fastapi import FastAPI, Request, HTTPException

SECRET = os.environ["XG_WEBHOOK_SECRET"].encode()
app = FastAPI()

def verify(body: bytes, signature: str) -> bool:
    expected = hmac.new(SECRET, body, hashlib.sha256).hexdigest()
    return hmac.compare_digest(expected, signature)

@app.post("/xg/tweet")
async def on_tweet(req: Request):
    body = await req.body()
    sig = req.headers.get("X-Signature", "")
    if not verify(body, sig):
        raise HTTPException(401, "bad signature")
    payload = await req.json()
    t = payload["data"]
    # Do something durable: enqueue, write to DB, fan out to Discord, etc.
    print(f"[@{t['author']}] {t['text']}")
    return {"ok": True}

Run with uvicorn app:app --port 8000 behind any HTTPS proxy. Xanguard signs each delivery with HMAC-SHA256 (hex) in the X-Signature header and sends the webhook ID in X-Webhook-Id. The payload is {"event": "tweet", "timestamp": ..., "data": {"tweet_id", "author", "text", "url", ...}}. Failed deliveries are retried up to 3 times (5xx and network errors), and a webhook is switched off after 10 consecutive failures. Return a 2xx quickly and do heavy work asynchronously.

Recipe 5 — A complete Telegram tweet bot in 50 lines

You can build a "tweet alerts to my private Telegram group" bot without running anything but a Telegram bot token and one Python file:

import asyncio, json, os, websockets, httpx, html

XG_KEY = os.environ["XG_KEY"]
TG_TOKEN = os.environ["TG_TOKEN"]
TG_CHAT = os.environ["TG_CHAT"]  # e.g. -1001234567890 for groups

TG_API = f"https://api.telegram.org/bot{TG_TOKEN}/sendMessage"

def fmt(handle: str, t: dict) -> str:
    h = html.escape
    head = f"<b>@{h(handle)}</b>"
    if t.get('type') in ('reply', 'quote'): head += f" ({t['type']})"
    link = f"https://x.com/{handle}/status/{t['id']}"  # the frame has no url field
    return f"{head}\n\n{h(t['text'])}\n\n<a href='{h(link)}'>open on x.com</a>"

async def main():
    url = "wss://api.xanguard.tech/v1/dt/realtime/ws"
    async with httpx.AsyncClient(timeout=10.0) as http:
        async with websockets.connect(url) as ws:
            await ws.recv()  # HELLO (op 10)
            await ws.send(json.dumps({"op": 2, "d": XG_KEY}))  # LOGIN (op 2) -> READY (op 4)
            async for raw in ws:
                msg = json.loads(raw)
                if msg.get("op") != 0: continue  # EVENT frames only
                d = msg["d"]
                if d.get("event") != "twitter.post.new": continue
                await http.post(TG_API, json={
                    "chat_id": TG_CHAT,
                    "text": fmt(d["task_info"]["handle"], d["data"]),
                    "parse_mode": "HTML",
                    "disable_web_page_preview": False,
                })

asyncio.run(main())

This is, in effect, a custom Telegram alerts product that you control end-to-end. You decide the formatting, the routing, the filtering. If you run it under a process supervisor with automatic restart, you have a production-grade bot for the cost of one VPS.

Recipe 6 — Convergence: fire when N KOLs converge on the same handle

This is the alpha-group killer feature. You want to know the instant three or more of your tracked accounts mention the same Twitter handle within a 10-minute window — that is often a coordinated narrative push or an organic alpha leak.

You can build this client-side in 30 lines on top of the WebSocket. For a related signal on X Communities, the hosted Convergence Tracker (/v1/ct/ws) fires when several tracked accounts join the same community.

import asyncio, json, os, re, time, collections, websockets

XG_KEY = os.environ["XG_KEY"]
WINDOW_SEC = 600
THRESHOLD = 3
MENTION_RE = re.compile(r"@([A-Za-z0-9_]{1,15})")

# mention -> deque of (timestamp, source_handle)
seen = collections.defaultdict(collections.deque)
fired = set()

def prune(dq, now):
    while dq and now - dq[0][0] > WINDOW_SEC:
        dq.popleft()

async def main():
    url = "wss://api.xanguard.tech/v1/dt/realtime/ws"
    async with websockets.connect(url) as ws:
        await ws.recv()  # HELLO (op 10)
        await ws.send(json.dumps({"op": 2, "d": XG_KEY}))  # LOGIN (op 2) -> READY (op 4)
        async for raw in ws:
            msg = json.loads(raw)
            if msg.get("op") != 0: continue  # EVENT frames only
            d = msg["d"]
            if d.get("event") != "twitter.post.new": continue
            t = d["data"]
            handle = d["task_info"]["handle"].lower()
            now = time.time()
            for m in MENTION_RE.findall(t["text"].lower()):
                dq = seen[m]
                prune(dq, now)
                if handle not in {s for _, s in dq}:
                    dq.append((now, handle))
                sources = {s for _, s in dq}
                if len(sources) >= THRESHOLD and m not in fired:
                    fired.add(m)
                    print(f"!! CONVERGENCE on @{m}: {sources}")

asyncio.run(main())

Run this against a list of crypto KOLs and you have a simple "narrative ignition" detector.

Production tips

Reconnect with backoff

WebSocket connections drop. Always wrap your consumer:

async def run_forever(handler):
    delay = 1
    while True:
        try:
            await handler()
            delay = 1
        except Exception as e:
            print(f"ws error: {e}, reconnecting in {delay}s")
            await asyncio.sleep(delay)
            delay = min(delay * 2, 30)

De-duplicate by tweet ID

Xanguard de-duplicates on the server side, but a twitter.post.update event reuses the event_id of the original post, and a reconnect can overlap. Keep a 5-minute LRU of data.id (or event_id) to avoid double-action on your end.

Verify HMAC on every webhook

If you use Tweet Alerts webhooks, reject any request whose signature does not match. Use hmac.compare_digest, not ==.

Persist your accounts list

Your tracked-accounts list lives in Xanguard. If your local code crashes, restarts, or migrates to a new server, your accounts and settings persist. Use the REST API as the source of truth, not your local config.

What else is available beyond the tweet feed

The /v1/dt/realtime/ws endpoint covers tweet alerts. Xanguard also offers feeds for adjacent signals, each with its own WebSocket:

Endpoint	What it streams	Typical use case
/v1/cw/ws	Community joins, renames, description changes, posts	Detect KOL coordination on private Twitter Communities
/v1/ct/ws	Convergence events (several tracked accounts join the same community)	Coordinated launch detection
/v1/dt/realtime/ws	Profile changes: name, bio, avatar, pinned tweet	Detect rebrands, KOL behavioral shifts
/v1/et/ws	Engagement velocity on specific tweets	Detect organic vs. paid virality
/v1/sa/ws	Keyword search matches across all of X (checked on a schedule, not real time)	Brand monitoring, ticker watch
/v1/trending/ws	Category-specific trending tweets	News routing, alpha by sector
/v1/pf/ws	Wallet activity (pump.fun launches, graduations)	On-chain → Twitter cross-signal pipelines

A B2B key gets you /v1/dt/realtime/ws; the other feeds are sold per product, each with its own API key from that product's bot. Profile changes on /v1/dt/realtime/ws need the Pro or Enterprise tier. See the API docs for each feed's protocol.

B2B pricing

The B2B plans are sized for developers building products on top of the API. Three tracks — Starter, Pro, Enterprise — each available at 50, 250, 500, or 1000 handles. All tracks include REST and WebSocket for the B2B feed.

Handles	Starter	Pro	Enterprise
50	$49	$99	$249
250	$229	$429	$979
500	$449	$749	$1,649
1000	$849	$1,349	$2,849

Starter streams tweets and deleted tweets. Pro adds profile changes and new followers. Enterprise adds follow/unfollow events and OCR of contract addresses in tweet images. For most builders, Starter 50 at $49 is the right place to begin. Upgrade in-place when you outgrow the handle count — there is no migration step.

Per-handle cost falls fast as you scale: $0.98/handle at Starter 50, $0.92 at Starter 250, $0.90 at Starter 500, $0.85 at Starter 1000. If you are running a TG bot, an alpha group, or any B2C product reselling Xanguard data, the higher-volume tiers are designed for you.

Next steps

1. Open @B2B_Xanguard_bot on Telegram and grab an API key.
2. Start with the Starter B2B tier — 50 tracked handles, full REST + WebSocket for $49/mo.
3. Scale up: Pro 50 ($99), Enterprise 50 ($249), or any of the higher-volume tiers up to 1,000 handles.
4. Build something. The fastest path to a working product is to take Recipe 5 (the Telegram bot) and modify it — change the formatter, add filters, add a database, add a trading hook.

If you build something interesting, tell us in the bot — we will feature it.

Frequently Asked Questions