Overview
TikTokLive provides several custom events beyond the standard protobuf events. These events give you more control over the connection lifecycle and allow you to handle unknown message types.
WebsocketResponseEvent
The WebsocketResponseEvent is triggered whenever any event is received from the WebSocket connection. This is useful for debugging or implementing custom handling logic for all incoming messages.
from TikTokLive import TikTokLiveClient
from TikTokLive.events import WebsocketResponseEvent
client = TikTokLiveClient(unique_id="@username")
@client.on(WebsocketResponseEvent)
async def on_any_event(event: WebsocketResponseEvent):
print(f"Received event: {event}")
if __name__ == '__main__':
client.run()
WebsocketResponseEvent inherits from both ProtoMessageFetchResult and BaseEvent, giving you access to raw protobuf data.
Connection Lifecycle Events
ConnectEvent
Triggered when the client successfully connects to a live stream and receives the first payload from the Sign Server.
from TikTokLive.events import ConnectEvent
@client.on(ConnectEvent)
async def on_connect(event: ConnectEvent):
print(f"Connected to @{event.unique_id}'s stream!")
print(f"Room ID: {event.room_id}")
unique_id (str): The TikTok usernameroom_id (int): The live room identifier
DisconnectEvent
Triggered when the client disconnects from a stream, either manually or due to an error.
from TikTokLive.events import DisconnectEvent
@client.on(DisconnectEvent)
async def on_disconnect(event: DisconnectEvent):
print("Disconnected from stream")
Stream State Events
LiveEndEvent
Triggered when the broadcaster ends their live stream.
from TikTokLive.events import LiveEndEvent
@client.on(LiveEndEvent)
async def on_live_end(event: LiveEndEvent):
print("Stream has ended")
LivePauseEvent
Triggered when the broadcaster pauses their stream.
from TikTokLive.events import LivePauseEvent
@client.on(LivePauseEvent)
async def on_pause(event: LivePauseEvent):
print("Stream paused")
LiveUnpauseEvent
Triggered when the broadcaster resumes a paused stream.
from TikTokLive.events import LiveUnpauseEvent
@client.on(LiveUnpauseEvent)
async def on_unpause(event: LiveUnpauseEvent):
print("Stream resumed")
Social Events
FollowEvent
A specialized version of SocialEvent for when a viewer follows the broadcaster.
from TikTokLive.events import FollowEvent
@client.on(FollowEvent)
async def on_follow(event: FollowEvent):
print(f"{event.user.nickname} followed!")
ShareEvent
A specialized version of SocialEvent for when a viewer shares the stream. Includes information about users who joined from the share.
from TikTokLive.events import ShareEvent
@client.on(ShareEvent)
async def on_share(event: ShareEvent):
print(f"{event.user.nickname} shared the stream!")
if event.users_joined:
print(f"{event.users_joined} users joined from this share")
users_joined (Optional[int]): Number of people who joined the stream from the share
SuperFanEvent
A specialized version of BarrageEvent for super fan messages.
from TikTokLive.events import SuperFanEvent
@client.on(SuperFanEvent)
async def on_superfan(event: SuperFanEvent):
print(f"Super fan message: {event.comment}")
Unknown Events
The UnknownEvent is triggered when TikTok sends a message type that is not yet tracked by the library.
from TikTokLive.events import UnknownEvent
@client.on(UnknownEvent)
async def on_unknown(event: UnknownEvent):
print(f"Unknown event type received: {type(event)}")
# You can inspect the raw protobuf data here
If you encounter UnknownEvent frequently, consider reporting it on the GitHub repository so support can be added. Complete Example
from TikTokLive import TikTokLiveClient
from TikTokLive.events import (
ConnectEvent,
DisconnectEvent,
LiveEndEvent,
LivePauseEvent,
LiveUnpauseEvent,
FollowEvent,
ShareEvent,
UnknownEvent,
WebsocketResponseEvent
)
client = TikTokLiveClient(unique_id="@username")
@client.on(ConnectEvent)
async def on_connect(event: ConnectEvent):
print(f"✓ Connected to @{event.unique_id}")
@client.on(DisconnectEvent)
async def on_disconnect(event: DisconnectEvent):
print("✗ Disconnected")
@client.on(LiveEndEvent)
async def on_live_end(event: LiveEndEvent):
print("Stream ended")
@client.on(LivePauseEvent)
async def on_pause(event: LivePauseEvent):
print("⏸ Stream paused")
@client.on(LiveUnpauseEvent)
async def on_unpause(event: LiveUnpauseEvent):
print("▶ Stream resumed")
@client.on(FollowEvent)
async def on_follow(event: FollowEvent):
print(f"👤 {event.user.nickname} followed")
@client.on(ShareEvent)
async def on_share(event: ShareEvent):
print(f"📤 {event.user.nickname} shared")
@client.on(UnknownEvent)
async def on_unknown(event: UnknownEvent):
print(f"❓ Unknown event: {type(event)}")
if __name__ == '__main__':
client.run()