Skip to main content

​
Connection Methods

​
run

def run(self, **kwargs) -> Task
Start a thread-blocking connection to TikTokLive. This is the simplest way to connect and will block the main thread until the connection ends.
​
**kwargs
dict
Keyword arguments passed to the start() method. See start for available options.
Returns: Task - The asyncio task once it’s finished Example: 
client = TikTokLiveClient(unique_id="@username")

if __name__ == "__main__":
    client.run()

​
connect

async def connect(
    self,
    callback: Optional[
        Union[
            Callable[[None], None],
            Callable[[None], Coroutine[None, None, None]],
            Coroutine[None, None, None],
        ]
    ] = None,
    **kwargs
) -> Task
Start a future-blocking connection to TikTokLive. This method is useful when you want to run the connection in an async context.
​
callback
Callable | Coroutine
default:"None"
An optional callback function or coroutine to run when connected. Can be sync or async.
​
**kwargs
dict
Keyword arguments passed to the start() method. See start for available options.
Returns: Task - The asyncio task once it’s finished Example: 
client = TikTokLiveClient(unique_id="@username")

async def on_connect():
    print("Connected to livestream!")

await client.connect(callback=on_connect)

​
start

async def start(
    self,
    *,
    process_connect_events: bool = True,
    compress_ws_events: bool = True,
    fetch_room_info: bool = False,
    fetch_gift_info: bool = False,
    fetch_live_check: bool = True,
    room_id: Optional[int] = None,
) -> Task
Create a non-blocking connection to TikTok LIVE and return the task. This is the underlying method called by run() and connect().
​
process_connect_events
bool
default:"True"
Whether to process initial events sent on room join (e.g., current viewers, gifts).
​
compress_ws_events
bool
default:"True"
Whether to compress WebSocket events using gzip compression. Should usually stay enabled for better performance.
​
fetch_room_info
bool
default:"False"
Whether to fetch room information on join. Room info will be available via the room_info property.
​
fetch_gift_info
bool
default:"False"
Whether to fetch gift information on join. Gift info will be available via the gift_info property.
​
fetch_live_check
bool
default:"True"
Whether to check if the user is live before connecting. You almost always want this enabled.
​
room_id
int
default:"None"
An override to connect directly to a livestream using its room ID. This skips HTML scraping and is useful for scaling, as scraping can result in TikTok blocks.
Returns: Task - The asyncio task containing the client heartbeat Raises:
  • AlreadyConnectedError - If the client is already connected
  • UserOfflineError - If fetch_live_check=True and the user is not live
  • UserNotFoundError - If the user does not exist
Example: 
client = TikTokLiveClient(unique_id="@username")

task = await client.start(
    fetch_room_info=True,
    fetch_gift_info=True
)

​
disconnect

async def disconnect(self, close_client: bool = False) -> None
Disconnect the client from the WebSocket connection.
​
close_client
bool
default:"False"
Whether to also close the HTTP client. Set to True if you don’t intend to reuse the client.
Returns: None Example: 
# Disconnect but keep HTTP client open for reuse
await client.disconnect()

# Disconnect and close everything
await client.disconnect(close_client=True)

​
close

async def close(self) -> None
Close the HTTP client and discard async sessions. Use this when you don’t intend to use the client again. Returns: None
This method is automatically called when disconnect(close_client=True) is used.
Example: 
# Close the client after disconnecting
await client.disconnect()
await client.close()

​
Event Listener Methods

​
has_listener

def has_listener(self, event: Type[Event]) -> bool
Check whether the client is listening to a given event type.
​
event
Type[Event]
required
The event class to check for listeners.
Returns: bool - True if the client has at least one listener for the event, False otherwise Example: 
from TikTokLive.events import CommentEvent, GiftEvent

# Check if listening to an event
if client.has_listener(CommentEvent):
    print("Listening to comments")

# Add listener and verify
client.add_listener(GiftEvent, on_gift)
assert client.has_listener(GiftEvent)  # True

​
on

def on(
    self,
    event: Type[Event],
    f: Optional[EventHandler] = None
) -> Union[Handler, Callable[[Handler], Handler]]
Decorator that registers a Python function as an event listener.
​
event
Type[Event]
required
The event class to listen for (e.g., CommentEvent, GiftEvent).
​
f
EventHandler
default:"None"
The function to handle the event. Can be sync or async.
Returns: Handler | Callable - The wrapped function as a pyee.Handler object Example: 
from TikTokLive.events import CommentEvent, GiftEvent

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

@client.on(GiftEvent)
async def on_gift(event: GiftEvent):
    print(f"{event.user.nickname} sent {event.gift.name}")

​
add_listener

def add_listener(
    self,
    event: Type[Event],
    f: EventHandler
) -> Handler
Method that registers a Python function as an event listener. This is the non-decorator version of on().
​
event
Type[Event]
required
The event class to listen for.
​
f
EventHandler
required
The function to handle the event.
Returns: Handler - The generated pyee.Handler object Example: 
from TikTokLive.events import CommentEvent

async def handle_comment(event: CommentEvent):
    print(event.comment)

client.add_listener(CommentEvent, handle_comment)

​
Utility Methods

​
is_live

async def is_live(self, unique_id: Optional[str | int] = None) -> bool
Check if a TikTok user is currently live.
​
unique_id
str | int
default:"None"
Optionally override the user to check. If not provided, checks the client’s configured user.
Returns: bool - Whether the user is currently live on TikTok Example: 
client = TikTokLiveClient(unique_id="@username")

# Check if the configured user is live
is_live = await client.is_live()

# Check if another user is live
is_another_live = await client.is_live("@another_user")

​
send_room_chat

async def send_room_chat(self, content: str) -> Any
Send a chat message to the room. Requires authentication.
​
content
str
required
The content of the chat message to send.
Returns: Any - The response from TikTok Example: 
response = await client.send_room_chat("Hello from TikTokLive!")

​
parse_unique_id

@classmethod
def parse_unique_id(cls, unique_id: str) -> str
Parse and clean a unique ID from a generic string. This method removes TikTok URL prefixes, /live suffixes, and leading @ symbols.
​
unique_id
str
required
The unique_id to parse and clean.
Returns: str - The parsed and cleaned unique_id Example: 
# All of these return "username"
TikTokLiveClient.parse_unique_id("@username")
TikTokLiveClient.parse_unique_id("https://www.tiktok.com/@username/live")
TikTokLiveClient.parse_unique_id("username")

Build docs developers (and LLMs) love

Get started for freeTalk to us