Build a TikTok Creator Alert Bot in Python (Uploads + Lives)

What you need

• Python 3.10+ and pip install websockets aiohttp
• A 1322 TikTok API key, a separate WebSocket key, and your WebSocket path, all from the dashboard API access page (plans from $200/mo, pricing; WebSocket delivery is a $100/mo add-on)
• A Discord webhook URL (channel settings → Integrations → Webhooks), or a Telegram bot token and chat id if you prefer Telegram

Read that first bullet twice, because it is the thing that trips people up. TikTok access issues two separate credentials, and they are not interchangeable. REST calls take the API key, as either X-API-Key: your_api_key orAuthorization: Bearer your_api_key. The WebSocket takes the WebSocket key, as either X-WS-Key: your_ws_key orAuthorization: Bearer your_ws_key. Mix them and you get a 401. Neither is accepted in the URL, so there is no?key= shortcut and no way to open the stream in a browser tab. The full contract is in the TikTok monitoring API reference.

Step 1: put creators on the tracked list

The tracked list is the entire configuration. Once a creator is on it, their events arrive on every delivery route you have enabled. One REST call per creator, API key in the header:

curl -X POST "https://tiktok.1322.io/v1/track" \
  -H "X-API-Key: YOUR_API_KEY" \
  -H "Content-Type: application/json" \
  -d '{"username": "examplecreator"}'

You can do the same thing from the dashboard or with the Discord bot if you would rather not script it. Either way it is the same list, and the bot below does not need to know anything about it.

Step 2: the bot

import asyncio, json, os
import aiohttp, websockets

WS_URL = os.environ["TIKTOK_WS_URL"]        # wss://tiktok.1322.io/ws/<opaque path>
WS_KEY = os.environ["TIKTOK_WS_KEY"]        # WebSocket key - NOT the REST API key
WEBHOOK = os.environ["DISCORD_WEBHOOK"]     # channel webhook URL

live_cards: dict[str, str] = {}   # room_id -> discord message id
seen: set[str] = set()            # event_id dedup across reconnects


async def post(session: aiohttp.ClientSession, text: str) -> str | None:
    # wait=true makes Discord return the created message, so we can edit it later
    async with session.post(WEBHOOK + "?wait=true", json={"content": text}) as r:
        return (await r.json()).get("id") if r.status < 300 else None


async def edit(session: aiohttp.ClientSession, message_id: str, text: str) -> None:
    await session.patch(WEBHOOK + "/messages/" + message_id, json={"content": text})


async def handle(session: aiohttp.ClientSession, e: dict) -> None:
    kind = e.get("type")
    who = (e.get("account") or {}).get("username", "?")
    obj = e.get("object") or {}

    if kind == "tiktok.upload.created":
        await post(session, f"[upload] @{who} posted: {obj.get('url')}")

    elif kind == "tiktok.repost.created":
        # original_author is an object, not a string
        original = (obj.get("original_author") or {}).get("username", "?")
        await post(session, f"[repost] @{who} reposted @{original}: {obj.get('url')}")

    elif kind == "tiktok.live.started":
        room = obj.get("room_id")
        message_id = await post(session, f"[LIVE] @{who} is live now")
        if room and message_id:
            live_cards[room] = message_id

    elif kind == "tiktok.live.ended":
        message_id = live_cards.pop(obj.get("room_id"), None)
        if message_id:
            await edit(session, message_id, f"[ended] @{who} finished the live session")
        else:
            await post(session, f"[ended] @{who} finished a live session")

    elif kind == "tiktok.media.ready":
        # NOTE: a media-ready frame carries no account field - it is a sibling of an
        # upload you have already seen, identified by the SAME event_id. Keep a
        # short event_id -> username map when you handle the upload if you want to
        # attribute it; the video_id is enough to fetch the file.
        media = obj.get("media_url") or next(iter(obj.get("image_urls") or []), None)
        if media:
            await post(session, f"[media] {obj.get('video_id')} file ready: {media}")


async def run() -> None:
    async with aiohttp.ClientSession() as session:
        while True:
            try:
                async with websockets.connect(
                    WS_URL, additional_headers={"Authorization": f"Bearer {WS_KEY}"}
                ) as ws:
                    print("connected")
                    async for raw in ws:
                        e = json.loads(raw)
                        event_id = e.get("event_id")
                        if event_id:
                            if event_id in seen:
                                continue
                            seen.add(event_id)
                        await handle(session, e)
            except Exception as exc:
                print(f"reconnecting ({exc})")
                await asyncio.sleep(2)


asyncio.run(run())

Run it: TIKTOK_WS_URL=... TIKTOK_WS_KEY=... DISCORD_WEBHOOK=... python bot.py

Step 3: why the live handling is the interesting part

Most tools that advertise live notifications fire once and forget, which is why so many Discord servers have a graveyard of "LIVE NOW" cards that stopped being true hours ago. The 1322 TikTok feed emitstiktok.live.started andtiktok.live.ended as two separate events carrying the same room_id. That shared id is what makes the open/close handling above work: the start handler keeps the Discord message id inlive_cards under the room id, and the end handler pops it back out and edits the card in place. One session, one message, accurate from open to close.

The ?wait=true on the Discord webhook POST is doing real work here too. Without it Discord returns 204 and no body, so you never learn the message id and can never edit it. With it you get the message object back, andPATCH /messages/<id> on the same webhook updates the card. The else branch covers the honest edge case: if your bot was down when the session started, there is no card to edit, so it posts a standalone note rather than swallowing the event.

Step 4: Telegram instead of Discord

Telegram supports the same open-and-close pattern, so only the two helpers and the DISCORD_WEBHOOK line change. Swap these in and the rest of the bot is untouched:

TG_TOKEN = os.environ["TG_TOKEN"]   # from @BotFather
TG_CHAT = os.environ["TG_CHAT"]     # target chat id
TG_API = "https://api.telegram.org/bot" + TG_TOKEN


async def post(session: aiohttp.ClientSession, text: str) -> str | None:
    async with session.post(TG_API + "/sendMessage",
                            json={"chat_id": TG_CHAT, "text": text}) as r:
        body = await r.json()
        return str((body.get("result") or {}).get("message_id") or "") or None


async def edit(session: aiohttp.ClientSession, message_id: str, text: str) -> None:
    await session.post(TG_API + "/editMessageText",
                       json={"chat_id": TG_CHAT, "message_id": int(message_id), "text": text})

How it works

• Dedup by event id. Every frame carries a unique event_id. The seen set means a reconnect cannot double-post an alert. In a long-running bot, bound that set or swap it for a TTL cache.
• Read the creator from the account object. account.username is who you are tracking. On a repost, object.original_author is an object with its own username, not a bare string, so a repost never gets misattributed to the creator who shared it.
• Wait for media-ready before you fetch. The playable file arrives as object.media_url on the tiktok.media.ready event, with object.image_urls for a photo carousel. Fetch on that event and you make one request instead of a retry loop against a file that does not exist yet.
• Reconnect, do not replay. The stream is real-time only and nothing is queued while you are disconnected, so the loop reconnects with a short backoff and carries on from live.

Where to take it next

• Route by creator: keep a dict of username → webhook so finance creators land in one channel and the brand watchlist in another, off the same socket.
• Filter on the description: the upload event carries the creator's text, so a keyword or ticker match can upgrade an alert to a mention instead of a quiet post.
• Ready-made minimal consumers for the platform feeds (X, Instagram, Truth Social, YouTube, Binance Square, News) live in our open-source examples repo: github.com/SisoSol/social-monitor-examples.
• Zero-code route: the included Discord bot does the posting and tracked-list management without a script. Custom code earns its keep when you want your own filters, formatting or destination, the same trade covered in alerts in Discord and Telegram.
• The commercial summary, if you are sending this to someone who does not write Python: get notified when a TikTok creator goes live.

FAQ