---

Three ways to use it

1. One-click (Windows)

Download start.bat, double-click. It installs the package + prompts for a username. Streams every chat, gift, like, follow in real-time to the console.

2. One-click (macOS / Linux)

curl -fsSL https://raw.githubusercontent.com/tiktool/tiktok-live-events/main/python/start.sh | bash

3. CLI

pip install tiktok-live-events
tiktok-live-events streamer_username

[ready]   connected to @streamer (room 7648...)
[chat]    fan123: love this!
[gift]    bigtipper -> Rose x99
[like]    casual_viewer (15)
[follow]  new_follower

4. Programmatic

import asyncio
from tiktok_live_events import TikTokLive

live = TikTokLive("streamer_username")

@live.on("chat")
def on_chat(e):
    print(f"{e['user']['uniqueId']}: {e['comment']}")

asyncio.run(live.run())

No key. No config. Just run it.

---

What you get

• Real-time chat, gifts, likes, follows, viewer counts, PK battles, TikTok native captions, gift catalog updates, moderation deletes, viewer entry-source analytics and the rest of the 90+ live event types.
• One WebSocket. Zero protocol code. No protobuf libraries. No headless browser.
• Full TypedDict typing for every event - Pyright + mypy autocomplete every field.
• Async first (asyncio). Auto-reconnect. Sync + async handlers both supported.
• Tiny. One runtime dependency (websockets).

The protocol decode happens on the TikTools edge. Your code only ever sees clean JSON.

---

How it works (anonymous / sandbox / paid)

The SDK has two connect strategies and picks one automatically based on your tier.

Direct mode (default for anonymous + sandbox)

Your machine opens the stream connection from your own network. The TikTools edge provides the connection details and turns raw frames into parsed JSON. You never run a protobuf library or a headless browser.

Managed mode (default for paid tiers)

You open ONE managed WebSocket to wss://api.tik.tools and receive parsed JSON events. Nothing else runs on your side.

Picking mode

Tier	Default mode	Override
Anonymous (no key)	direct	mode='managed'
Sandbox (7-day evaluation, no credit card; 5,000 requests/day, 3 streams)	direct	mode='managed'
Basic+ (paid)	managed	mode='direct'

The default is mode='auto' - the SDK asks our edge which mode fits your tier and picks the right one. Force a mode explicitly when you want to override.

live = TikTokLive('streamer', api_key='...', mode='direct')   # force direct
live = TikTokLive('streamer', api_key='...', mode='managed')  # force managed
live = TikTokLive('streamer')                                 # auto (default)

Pricing tiers + current per-mode caps live on the pricing page.

Server address (paid plans)

Paid plans (Basic and up) can connect through the direct endpoint https://direct.tik.tools with the same API key. Set base_url once; the WebSocket address (wss://direct.tik.tools) is derived from it.

live = TikTokLive('streamer', api_key='...', base_url='https://direct.tik.tools')

Or set TIKTOOL_BASE_URL=https://direct.tik.tools in the environment. This is the server address only and is separate from mode. Sandbox keys and anonymous use are refused on the direct endpoint. Without the option or the variable, the SDK uses https://api.tik.tools as before.

---

CLI reference

tiktok-live-events <username> [options]

Options:
  -f, --filter <list>     Comma-separated event types (default: all)
                          e.g. chat,gift,follow,viewer,like
      --json              Emit each event as one JSON line (machine-readable)
  -h, --help              Show this help

Examples:
  tiktok-live-events streamer
  tiktok-live-events streamer --filter chat,gift
  tiktok-live-events streamer --json > events.ndjson

---

Install

pip install tiktok-live-events

# uv / poetry / pipx
uv add tiktok-live-events
poetry add tiktok-live-events
pipx install tiktok-live-events

---

Quick start (SDK)

import asyncio
from tiktok_live_events import TikTokLive

live = TikTokLive("streamer_username")

@live.on("connected")
def on_connected(_):
    print("Connected.")

@live.on("chat")
def on_chat(e):
    print(f"{e['user']['uniqueId']}: {e['comment']}")

@live.on("gift")
def on_gift(e):
    print(f"{e['user']['uniqueId']} sent {e['giftName']} x{e['repeatCount']} ({e['diamondCount']} diamonds)")

@live.on("like")
def on_like(e):
    print(f"{e['likeCount']} likes (total: {e['totalLikes']})")

asyncio.run(live.run())

---

Events

Every event is dispatched by name. Handlers receive a dict matching the event's TypedDict (ChatEvent, GiftEvent, BattleArmiesEvent, ...). Each payload extends BaseEvent (type, timestamp, msgId, optional protoVersion).

Core live events

Event	What it carries
connected	Socket open.
disconnected	Socket close.
roomInfo	One-shot post-connect: {roomId, wsHost, clusterRegion, connectedAt}.
chat	user, comment, emotes, optional starred. v3 adds language (auto-detected), messageUuid, replyToUser (~8% of chats are replies).
gift	giftId, giftName, diamondCount, repeatCount, repeatEnd, giftType. v3 adds transactionId, senderUserId, relationship (joinDayNumber).
like	likeCount (this batch), totalLikes (room cumulative).
member	Viewer joined. v3 adds entrySource ("homepage_hot-live_cell", "follow-tab", ...), entryAction ("draw"/"click"), entryType ("rec").
social	Follow / share.
roomUserSeq	Periodic viewer count tick. Also emitted as viewer_count (alias, same payload).
subscribe	A viewer subscribed.

PK / battle events

Event	What it carries
battle	PK lifecycle. status (1=ACTIVE, 2=STARTING, 3=ENDED, 4=PREPARING), battleDuration, teams. v3 adds extraHostUserIds, layoutSubtype.
battleArmies	Per-host MVP breakdown. hosts[].contributors[] sorted MVP first. v3 adds transactionId.
battleItemCard	Booster card: x2 / x3 multipliers, gloves (crit), mist, thunder, extra-time, match-guide. Carries TikTok CDN overlay assets.
battlePunishFinish	Loser-side punishment screen ended.
battleNotice, battleGameplay	PK notice + mini-game state.
linkLayer, linkMicOpponentGift, linkScreenChange, cohostLayoutUpdate	Link-mic negotiation, opponent-side gifts, layout flips.
competition, competitionContributor, guestShowdown	Cross-stream competitions + guest showdowns.

Native captions (v3)

Event	What it carries
caption	NEW in v3. TikTok native auto-captions on the LIVE WebSocket. text, language (auto-detected), isFinal, startedAtMs, endsAtMs.

Creator + room

Event	What it carries
goalUpdate	Stream goal progress (subscriber, gift, watch-time goals).
commentTray, roomPin, roomSticker, inRoomBanner, bottomMessage	Room UI events.
hostBoard, rankText, rankUpdate, hourlyRank	Leaderboard / rank events.
privilegeAdvance, accessRecall, roomVerify	Viewer privilege + content-classification events.
anchorToolModification, streamStatus, shareRevenueNotice	Creator-side metadata flips.
capsule, hotRoom, linkMicAnchorGuide	TikTok host nudges.

Moderation / safety

Event	What it carries
imDelete	Chat moderation delete. Correlate via chat.messageUuid (v3).
unauthorizedMember	Non-logged-in viewer hit a gated feature.
barrage	Raw barrage feed.
superFan, superFanJoin, superFanBox	Super-fan lifecycle.
emoteChat	Inline emote message.

Gift catalog + ecommerce

Event	What it carries
giftPanelUpdate	Real-time gift catalog change.
giftCollectionUpdate	Host curated gift set changed. v3
giftDynamicRestriction, giftGallery, giftUnlock, viewerPicksUpdate	Gift availability flips, host gift wall, gated-gift reveals, viewer-pick highlights.
oecLiveShopping, oecLiveManager, oecLiveBillboard, ecShortItemRefresh	OEC live-shopping events.

Engagement + AI

Event	What it carries
aiSummary	TikTok AI summary of the room (entry-time recap, multi-language).
poll, shortTouch	In-stream poll lifecycle.
question, questionSelected, questionSlideDown	Q&A round events.
pictionaryStart, pictionaryUpdate, pictionaryEnd, pictionaryExit	Drawing-game rounds. v3
karaokeReq	Viewer queued / requested a track on the host's karaoke widget. v3
subPin	Comment pinned by a paid subscriber via the sub-only pin slot. v3
toast, gapHighlightPushGuide	Generic toast popups and first-render UX hints. v3
gameAutoPostNotice	Notice posted automatically by an in-room mini-game. v3
cohostSettingsUpdate	Cohost settings updated (slot count, layout, permissions). v3
fansEvent, fanTicket	Fan-club events.
envelope, envelopePortal	Red-envelope drops + multi-room portal chain.
gameMoment, gameServerFeature	TikTok Gaming live integration.
groupLiveMemberNotify	Group-live member join / leave.
perception	Perception event (mute cancel, hint signal).
control, room, liveIntro	Stream control + room metadata.

Universal field: extras

Every event carries an optional extras: dict[str, ...] map containing any payload field TikTok ships that doesn't yet have a typed name. New fields appear automatically the day TikTok introduces them - no SDK upgrade required. Use it as a forward-compat hook:

@live.on('chat')
def on_chat(e):
    if e.get('extras', {}).get('18'):
        print('chat flag 18:', e['extras']['18'])

Catch-all

• event - Fires once for every decoded event (dump-to-queue pattern).
• unknown - Fires when TikTok ships a method not yet modelled (forward-compat hook).

---

Recipes

Chat logger

import asyncio
from tiktok_live_events import TikTokLive

live = TikTokLive("creator")

@live.on("chat")
def on_chat(e):
    print(f"{e['user']['uniqueId']}: {e['comment']}")

asyncio.run(live.run())

Gift leaderboard

import asyncio
from collections import defaultdict
from tiktok_live_events import TikTokLive

live = TikTokLive("creator")
board = defaultdict(int)

@live.on("gift")
def on_gift(e):
    if not e.get("repeatEnd"):
        return
    board[e["user"]["uniqueId"]] += e["diamondCount"] * e["repeatCount"]

async def print_top():
    while True:
        await asyncio.sleep(5)
        top = sorted(board.items(), key=lambda x: -x[1])[:10]
        print("\n-- TOP GIFTERS --")
        for i, (user, diamonds) in enumerate(top, 1):
            print(f"{i:>2}. {user:<20} {diamonds} diamonds")

async def main():
    asyncio.create_task(print_top())
    await live.run()

asyncio.run(main())

PK MVP tracker

import asyncio
from tiktok_live_events import TikTokLive

live = TikTokLive("creator")

@live.on("battle")
def on_battle(e):
    print(f"[battle] status={e['status']} id={e['battleId']} duration={e['battleDuration']}s")

@live.on("battleArmies")
def on_armies(e):
    print(f"[armies] remaining={e.get('secsRemaining')}s")
    for host in e.get("hosts", []):
        print(f"  Host {host['hostUserId']} total={host['teamTotalScore']}")
        mvp = host.get("contributors", [None])[0]
        if mvp:
            print(f"    MVP {mvp['nickname']} {mvp['score']} diamonds")

@live.on("battleItemCard")
def on_card(e):
    if e.get("multiplier", 0) > 0:
        print(f"[card] x{e['multiplier']} booster from {e['senderNickname']}")
    else:
        print(f"[card] effect={e['effect']} from {e['senderNickname']} duration={e['durationSec']}s")

asyncio.run(live.run())

TikTok native captions to file

import asyncio
from tiktok_live_events import TikTokLive

live = TikTokLive("creator")

@live.on("caption")
def on_caption(e):
    if e.get("isFinal"):
        with open("transcript.txt", "a", encoding="utf-8") as f:
            f.write(e["text"] + "\n")

asyncio.run(live.run())

Discord webhook relay

import asyncio, json
import urllib.request
from tiktok_live_events import TikTokLive

WEBHOOK = "https://discord.com/api/webhooks/..."
live = TikTokLive("creator")

def send(content):
    body = json.dumps({"content": content}).encode("utf-8")
    req = urllib.request.Request(WEBHOOK, data=body, headers={"content-type": "application/json"})
    try: urllib.request.urlopen(req, timeout=4).read()
    except Exception: pass

@live.on("chat")
def on_chat(e):
    send(f"**{e['user']['uniqueId']}**: {e['comment']}")

@live.on("gift")
def on_gift(e):
    if e.get("repeatEnd"):
        send(f":gift: {e['user']['uniqueId']} sent {e['giftName']} x{e['repeatCount']} ({e['diamondCount']} diamonds)")

asyncio.run(live.run())

More ready-to-run recipes in examples/.

---

API reference

TikTokLive(unique_id, *, auto_reconnect=True, max_reconnect_attempts=5, debug=False)

Construct a client. unique_id is the TikTok @username (with or without @).

Param	Type	Default	Description
unique_id	str	-	TikTok username.
auto_reconnect	bool	True	Reconnect with exponential backoff if the socket drops.
max_reconnect_attempts	int	5	Stop after N reconnect attempts.
debug	bool	False	Verbose logging via the tiktok_live_events logger.

live.on(event)

Decorator. Register a sync or async handler.

@live.on("chat")
def handler(e): ...

@live.on("gift")
async def handler(e): ...

Pass "event" to receive every decoded event in a single handler.

await live.run()

Connect and pump events until stop() is called or the reconnect budget is exhausted.

live.stop()

Signal run() to exit on the next iteration.

live.connected

bool - whether the socket is currently open.

---

Compatibility

• Python >= 3.9 (uses asyncio, websockets, type hints).
• Works on Windows, macOS, Linux, Docker, serverless.
• Tested against asyncio + uvloop.

---

Powered by

This package connects to the TikTools edge. Schema, decoding and protocol updates are handled server-side - your pip install never needs to bump when TikTok ships a wire change.

---

License

MIT

This is an independent third-party project. Not affiliated with, endorsed by, or in any way officially connected to TikTok or ByteDance Ltd. "TikTok" is a trademark of ByteDance Ltd; the name appears here for search discoverability.