A Python library to connect to and read events from TikTok's LIVE service.

A Python library to receive and decode livestream events such as comments and gifts in real-time from TikTok's LIVE service by connecting to TikTok's internal Webcast service. This library includes a wrapper that connects to the Webcast service using only a user's unique_id and allows you to join your livestream as well as that of other streamers. No credentials are required to use TikTokLive.

Join the support discord and visit the #py-support channel for questions, contributions and ideas. Feel free to make pull requests with missing/new features, fixes, etc.

UPDATE:
Due to a change on the part of TikTok, versions prior to v4.3.8 are no longer functional. If you are using an unsupported version, upgrade to the latest version using the pip install TikTokLive --upgrade command.

TikTok LIVE is available in several alternative languages with only slight differences between the versions:

• Node.JS: https://github.com/zerodytrash/TikTok-Live-Connector
• Go: https://github.com/Davincible/gotiktoklive
• C#/Unity: https://github.com/frankvHoof93/TikTokLiveSharp
• Java: https://github.com/jwdeveloper/TikTok-Live-Java

Primary Information

• Documentation
• Contributors
• License
• Quickstart Examples

Resources & Guides

1. David's Intro Tutorial
2. Getting Started
3. Params & Options
4. Client Methods
5. Client Attributes
6. TikTok Events

I cannot recommend this tutorial enough for people trying to get started. It is succinct, informative and easy to understand, created by David Teather, the creator of the Python TikTok-Api package. Click the thumbnail to warp.

1. Install the module via pip

> pip install TikTokLive

2. Create your first chat connection

from TikTokLive import TikTokLiveClient
from TikTokLive.types.events import CommentEvent, ConnectEvent

# Instantiate the client with the user's username
client: TikTokLiveClient = TikTokLiveClient(unique_id="@isaackogz")


# Define how you want to handle specific events via decorator
@client.on("connect")
async def on_connect(_: ConnectEvent):
    print("Connected to Room ID:", client.room_id)


# Notice no decorator?
async def on_comment(event: CommentEvent):
    print(f"{event.user.nickname} -> {event.comment}")


# Define handling an event via a "callback"
client.add_listener("comment", on_comment)

if __name__ == '__main__':
    # Run the client and block the main thread
    # await client.start() to run non-blocking
    client.run()

For more examples, see the examples folder provided in the tree.

To create a new TikTokLiveClient object the following parameter is required. You can optionally add configuration options to this via kwargs.

TikTokLiveClient(unique_id, **options)

A TikTokLiveClient object contains the following methods:

A TikTokLiveClient object contains the following attributes:

A TikTokLiveClient object has the following events. You can add events either by doing client.add_listener("event_name", callable) or by decorating a function with @client.on("event_name") that includes an event payload parameter.

Triggered when the websocket connection is successfully established.

@client.on("connect")
async def on_connect(event: ConnectEvent):
    print("Connected")

Triggered when the connection is terminated. You can call start() to reconnect . Note that you should wait a little bit before attempting a reconnect to to avoid being rate-limited.

@client.on("disconnect")
async def on_disconnect(event: DisconnectEvent):
    print("Disconnected")

Triggered every time someone likes the stream.

@client.on("like")
async def on_like(event: LikeEvent):
    print(f"@{event.user.unique_id} liked the stream!")

Triggered every time a new person joins the stream.

@client.on("join")
async def on_join(event: JoinEvent):
    print(f"@{event.user.unique_id} joined the stream!")

Triggered every time a gift arrives. Extra information can be gleamed off the available_gifts client attribute.

NOTE: Users have the capability to send gifts in a streak. This increases the event.gift.count value until the user terminates the streak. During this time new gift events are triggered again and again with an increased event.gift.count value. It should be noted that after the end of the streak, another gift event is triggered, which signals the end of the streak via event.gift.is_repeating:1. This applies only to gifts with event.gift.info.type:1. This means that even if the user sends an event.gift.info.type:1 gift only once, you may receive the event twice. Once with event.gift.is_repeating:0 and once with event.gift.is_repeating:1. Therefore, the event should be handled as follows in one of TWO ways. These are the same, except the second is a 'higher level' implementation using TikTokLive API features:

@client.on("gift")
async def on_gift(event: GiftEvent):
    # If it's type 1 and the streak is over
    if event.gift.info.type == 1:
        if event.gift.is_repeating == 1:
            print(f"{event.user.unique_id} sent {event.gift.count}x \"{event.gift.info.name}\"")

    # It's not type 1, which means it can't have a streak & is automatically over
    elif event.gift.info.type != 1:
        print(f"{event.user.unique_id} sent \"{event.gift.info.name}\"")

@client.on("gift")
async def on_gift(event: GiftEvent):
    # Streakable gift & streak is over
    if event.gift.streakable and not event.gift.streaking:
        print(f"{event.user.unique_id} sent {event.gift.count}x \"{event.gift.info.name}\"")

    # Non-streakable gift
    elif not event.gift.streakable:
        print(f"{event.user.unique_id} sent \"{event.gift.info.name}\"")

Triggered every time someone follows the streamer.

@client.on("follow")
async def on_follow(event: FollowEvent):
    print(f"@{event.user.unique_id} followed the streamer!")

Triggered every time someone shares the stream.

@client.on("share")
async def on_share(event: ShareEvent):
    print(f"@{event.user.unique_id} shared the stream!")

Triggered when 5 or 10 users join from a viewer's share link.

@client.on("more_share")
async def on_connect(event: MoreShareEvent):
    print(f"More than {event.amount} users have joined from {event.user.unique_id}'s share link!")

Triggered every time the viewer count is updated. This event also updates the cached viewer count by default.

@client.on("viewer_update")
async def on_connect(event: ViewerUpdateEvent):

    # Viewer Count
    print("Received a new viewer count:", event.viewer_count)
    print("The client automatically sets the count as an attribute too:", client.viewer_count)
    
    # Top VIewers
    print("You can even get the top viewers (by coins gifted)!:", event.top_viewers)
    print("The client automatically sets the top viewers as an attribute too:", client.top_viewers)

Triggered every time someone comments on the live.

NOTE: Some comments will be missing. Certain "low quality" comments will ONLY show up when a session_id is passed to the client.

@client.on("comment")
async def on_connect(event: CommentEvent):
    print(f"{event.user.nickname} -> {event.comment}")

Triggered when someone sends a subscription emote comment to the live chat.

@client.on("emote")
async def on_connect(event: EmoteEvent):
    print(f"{event.user.nickname} -> {event.emote.image.url}")

Triggered when someone sends an envelope (treasure box) to the TikTok streamer.

@client.on("envelope")
async def on_connect(event: EnvelopeEvent):
    print(f"{event.treasure_box_user.unique_id} -> {event.treasure_box_data}")

Triggered when a stream rank update is sent out. Can be Weekly Ranking or Rising Star!

@client.on("ranking_update")
async def on_connect(event: RankingUpdateEvent):
    print(f"{event.user.unique_id} has the rank #{event.rank} for the {event.type} leaderboard.")

Triggered when a user rank update is sent out. Can be Top Viewer status.

@client.on("user_ranking_update")
async def on_connect(event: UserRankingUpdateEvent):
    print(f"{event.user.unique_id} just became a #{event.rank} top viewer!")

Triggered when a Mic Battle starts!

@client.on("mic_battle_start")
async def on_connect(event: MicBattleStartEvent):
    print(f"A Mic battle has started!")

Triggered when information is received about a mic battle's progress.

@client.on("mic_battle_update")
async def on_connect(event: MicBattleUpdateEvent):
    print(f"An army in the mic battle has received points, or the status of the battle has changed!")

Triggered when the live stream gets terminated by the host.

@client.on("live_end")
async def on_connect(event: LiveEndEvent):
    print(f"Livestream ended :(")

Triggered when an intro message is sent to the live room. An intro message is basically a pinned message at the top of chat when you join a room.

This event only fires if "process_initial_data" is enabled and the streamer has an intro message configured.

@client.on("intro_message")
async def on_connect(event: IntroMessageEvent):
    print(f"Message: {event.message}")

Triggered when ANY unknown event is received that is not yet handled by this client.

This includes both events where the protobuf has NOT been decoded, as well as events where it has been decoded, but no event object has been created (e.g. it's useless data).

Use this event to debug and find new events to add to TikTokLive. Mention them here when you do.

This event is very advanced and handles both types of cases, an API to help you decode including offering the binary as base64. You can plug base64 into https://protobuf-decoder.netlify.app/ to reverse-engineer the protobuf schema.

@client.on("unknown")
async def on_connect(event: UnknownEvent):
    print(f"Event Type: {event.type}")
    print(f"Event Base64: {event.base64}")

Triggered when there is an error in the client or error handlers.

If this handler is not present in the code, an internal default handler will log errors in the console. If a handler is added, all error handling (including logging) is up to the individual.

Warning: If you listen for the error event and do not log errors, you will not see when an error occurs. This expected behaviour, listening to the error event overrides & disables the built-in one.

@client.on("error")
async def on_connect(error: Exception):
    # Handle the error
    if isinstance(error, SomeRandomError):
        print("Handle some error!")
        return

    # Otherwise, log the error
    # You can use the internal method, but ideally your own
    client._log_error(error)

• Isaac Kogan - Creator, Primary Maintainer, and Reverse-Engineering - isaackogan
• Zerody - Initial Reverse-Engineering Protobuf & Support - Zerody
• Davincible - Reverse-Engineering Stream Downloads - davincible
• David Teather - TikTokLive Introduction Tutorial - davidteather

See also the full list of contributors who have participated in this project.

This project is licensed under the MIT License - see the LICENSE file for details.