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 webhook receiver that classifies tweets by contract address, and a multi-product consumer that joins tweet alerts with profile-change events.

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

One API key, three transport options:

• REST at https://api.xanguard.tech/v1/* — manage tracked accounts, settings, webhooks, and subscription state.
• WebSocket at wss://api.xanguard.tech/v1/ws — receive tweet alerts as they happen. Median delivery latency is ~380 ms end-to-end (measured over the last 24 hours, 1,767 tracked handles, 144,251 tweets).
• Webhooks — HMAC-signed HTTPS POSTs to any endpoint you control. Built-in retries, circuit breaker, and signature verification.

Every paid tier includes all three. The differences between tiers are the number of tracked accounts and the request rate limit (2 to 100 requests/second).

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, then /subscribe to pick a plan (Starter tier is enough for everything in this tutorial; the Free consumer tier on @Xanguard_bot does not include API access).
3. Send /apikey to generate a key. It looks like xg_live_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:

curl -X POST https://api.xanguard.tech/v1/accounts \
  -H "Authorization: Bearer $XG_KEY" \
  -H "Content-Type: application/json" \
  -d '{"handles":["elonmusk","VitalikButerin","cz_binance"]}'

Response:

{
  "ok": true,
  "data": {
    "added": ["elonmusk", "vitalikbuterin", "cz_binance"],
    "already_exists": [],
    "total": 3,
    "limit": 75
  }
}

Stream tweets:

import asyncio, json, os, websockets

XG_KEY = os.environ["XG_KEY"]

async def main():
    url = "wss://api.xanguard.tech/v1/ws"
    headers = {"Authorization": f"Bearer {XG_KEY}"}
    async with websockets.connect(url, extra_headers=headers) as ws:
        async for raw in ws:
            msg = json.loads(raw)
            if msg.get("type") == "alert":
                t = msg["data"]
                print(f"[@{t['handle']}] {t['text'][:140]}")
                print(f"  {t['url']}\n")

asyncio.run(main())

Run it. The next time Elon, Vitalik, or CZ posts, you will see the tweet within ~0.4 seconds.

Recipe 2 — Filter by keyword and contract address

Most use cases do not want every tweet from every tracked account. Two filtering knobs are available:

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 the most popular filter for trading bots.

# Only deliver Vitalik tweets that mention "ethereum" or "L2"
curl -X PUT https://api.xanguard.tech/v1/accounts/vitalikbuterin/keywords \
  -H "Authorization: Bearer $XG_KEY" \
  -H "Content-Type: application/json" \
  -d '{"keywords":["ethereum","L2"]}'

# Globally suppress everything except contract-address tweets
curl -X PATCH https://api.xanguard.tech/v1/settings \
  -H "Authorization: Bearer $XG_KEY" \
  -H "Content-Type: application/json" \
  -d '{"contracts_only":true}'

From here, your WebSocket only fires on tweets containing a contract address from any tracked handle. Three lines of code, and you have a KOL contract-call firehose.

Recipe 3 — A complete Solana sniper bridge

This is what 80% of our API users build first: 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"]  # e.g. your Photon/BullX/Trojan webhook

# 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/ws"
    headers = {"Authorization": f"Bearer {XG_KEY}"}
    async with websockets.connect(url, extra_headers=headers) as ws:
        async for raw in ws:
            msg = json.loads(raw)
            if msg.get("type") != "alert": continue
            t = msg["data"]
            cas = extract_solana_cas(t["text"])
            if not cas: continue
            for ca in cas:
                print(f"[@{t['handle']}] CA: {ca}")
                await execute_buy(ca, t["handle"], t["url"])

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 the API itself, which is already optimized.

Recipe 4 — Webhook receiver (server-side delivery)

WebSockets are best for low-latency clients that stay connected. Webhooks are better for serverless, durable, or load-balanced deployments. Xanguard signs every webhook with HMAC-SHA256 so you can verify authenticity.

Register the webhook:

curl -X POST https://api.xanguard.tech/v1/webhooks \
  -H "Authorization: Bearer $XG_KEY" \
  -H "Content-Type: application/json" \
  -d '{
    "url": "https://yourdomain.com/xg/tweet",
    "events": ["tweet"],
    "filter_handles": ["elonmusk","cz_binance"]
  }'

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 retries 5xx responses three times with exponential backoff (1s, 2s, 4s); 4xx is treated as misconfiguration and fails immediately. After 10 consecutive failures the webhook is auto-disabled to protect your endpoint.

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(t: dict) -> str:
    h = html.escape
    head = f"<b>@{h(t['handle'])}</b>"
    if t['is_reply']: head += " (reply)"
    if t['is_quote']: head += " (quote)"
    return f"{head}\n\n{h(t['text'])}\n\n<a href='{h(t['url'])}'>open on x.com</a>"

async def main():
    url = "wss://api.xanguard.tech/v1/ws"
    headers = {"Authorization": f"Bearer {XG_KEY}"}
    async with httpx.AsyncClient(timeout=10.0) as http:
        async with websockets.connect(url, extra_headers=headers) as ws:
            async for raw in ws:
                msg = json.loads(raw)
                if msg.get("type") != "alert": continue
                t = msg["data"]
                await http.post(TG_API, json={
                    "chat_id": TG_CHAT,
                    "text": fmt(t),
                    "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 wrap it in a Docker container with a restart policy, 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. Xanguard also exposes a hosted version at /v1/ct/ws if you would rather subscribe to pre-computed convergence events; both work.

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/ws"
    headers = {"Authorization": f"Bearer {XG_KEY}"}
    async with websockets.connect(url, extra_headers=headers) as ws:
        async for raw in ws:
            msg = json.loads(raw)
            if msg.get("type") != "alert": continue
            t = msg["data"]
            now = time.time()
            for m in MENTION_RE.findall(t["text"].lower()):
                dq = seen[m]
                prune(dq, now)
                if t["handle"].lower() not in {s for _, s in dq}:
                    dq.append((now, t["handle"].lower()))
                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 the 500 most-followed crypto KOLs and you have a "narrative ignition" detector that beats most paid alpha groups.

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 if you process the same stream through both WebSocket and webhook (for redundancy), keep a 5-minute LRU of tweet_id to avoid double-action on your end.

Verify HMAC on every webhook

An unsigned webhook is a SQL-injection vector waiting to happen. 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/ws endpoint covers tweet alerts. Xanguard also offers specialized real-time feeds for adjacent signals, each with its own WebSocket and webhook:

Endpoint	What it streams	Typical use case
/v1/cw/ws	Community joins, leaves, posts	Detect KOL coordination on private Twitter Communities
/v1/ct/ws	Convergence events (N handles → 1 target)	Narrative ignition, coordinated shilling 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-driven search alerts (cross-handle)	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

Each is a tier-gated add-on. The base Tweet Alerts subscription gets you /v1/ws; the others are sold per product. The auth model is identical: same API key, same WebSocket protocol, different data.

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, 200, 500, 1000, or 2500 handles. All tracks include REST + WebSocket + Webhooks on every endpoint listed in the table above.

Tier	Handles	Starter	Pro	Enterprise
Entry	50	$49	$99	$249
Standard	200	$149	$299	$749
Scale	500	$299	$599	$1,499
High volume	1000	$499	$999	$2,499
Bulk	2500	$999	—	—

Differences between Starter, Pro, and Enterprise are SLA, dedicated support, and burst rate-limit headroom. 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.50 at Starter 1000, $0.40 at Starter 2500. 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 + WS + webhooks for $49/mo.
3. Scale up: Pro 50 ($99), Enterprise 50 ($249), or any of the higher-volume tiers up to 2,500 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.