Event Handling

Overview

TikTokLive provides two methods for handling events:

Decorator syntax with @client.on() - Recommended for most use cases
Programmatic registration with client.add_listener() - Useful for dynamic event handling

Both methods work identically; choose the one that fits your coding style.

Using Decorators

Basic Decorator Usage

The @client.on() decorator is the most common way to handle events:

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

client = TikTokLiveClient(unique_id="@username")

@client.on(ConnectEvent)
async def on_connect(event: ConnectEvent):
    print(f"Connected to @{event.unique_id}!")

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

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

Multiple Event Handlers

You can register multiple handlers for the same event:

@client.on(CommentEvent)
async def log_comment(event: CommentEvent):
    logger.info(f"Comment from {event.user_info.nickname}")

@client.on(CommentEvent)
async def process_comment(event: CommentEvent):
    # Process the comment for commands, etc.
    if event.comment.startswith("!"):
        await handle_command(event)

Both handlers will be called for each CommentEvent. They execute in the order they were registered.

Decorator Method Signature

def on(
    event: Type[Event],
    f: Optional[EventHandler] = None
) -> Union[Handler, Callable[[Handler], Handler]]

event

Type[Event]

required

The event class to listen for (e.g., CommentEvent, GiftEvent).

EventHandler

default:"None"

Optional function to use as handler (used internally).

Returns: A pyee.Handler object or decorator function.

Using add_listener()

Basic add_listener Usage

The add_listener() method allows you to register event handlers programmatically:

from TikTokLive import TikTokLiveClient
from TikTokLive.events import CommentEvent, GiftEvent

client = TikTokLiveClient(unique_id="@username")

async def handle_comment(event: CommentEvent):
    print(f"{event.user_info.nickname}: {event.comment}")

async def handle_gift(event: GiftEvent):
    if not event.streaking:
        print(f"Gift received: {event.gift.name}")

# Register handlers
client.add_listener(CommentEvent, handle_comment)
client.add_listener(GiftEvent, handle_gift)

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

Dynamic Event Registration

Use add_listener() when you need to register handlers conditionally or at runtime:

def setup_event_handlers(client: TikTokLiveClient, config: dict):
    """Dynamically set up event handlers based on configuration"""
    
    if config.get("log_comments"):
        client.add_listener(CommentEvent, log_comment_handler)
    
    if config.get("track_gifts"):
        client.add_listener(GiftEvent, track_gift_handler)
    
    if config.get("monitor_joins"):
        client.add_listener(JoinEvent, monitor_join_handler)

# Use it
config = {"log_comments": True, "track_gifts": True}
client = TikTokLiveClient(unique_id="@username")
setup_event_handlers(client, config)

add_listener Method Signature

def add_listener(
    event: Type[Event],
    f: EventHandler
) -> Handler

event

Type[Event]

required

The event class to listen for.

EventHandler

required

The async function to call when the event occurs.

Returns: A pyee.Handler object representing the registered listener.

Event Handler Functions

Async Handlers

All event handlers must be async functions:

# ✅ Correct - async function
@client.on(CommentEvent)
async def on_comment(event: CommentEvent):
    await process_comment(event)

# ❌ Wrong - sync function won't work
@client.on(CommentEvent)
def on_comment(event: CommentEvent):  # Missing async!
    print(event.comment)

Handler Parameters

Event handlers receive exactly one parameter: the event object.

@client.on(GiftEvent)
async def on_gift(event: GiftEvent):
    # Access event properties
    sender = event.from_user.nickname
    gift_name = event.gift.name
    repeat = event.repeat_count
    
    print(f"{sender} sent {repeat}x {gift_name}")

Type Hints

Always use type hints for better IDE support and code clarity:

from TikTokLive.events import CommentEvent, GiftEvent, JoinEvent

@client.on(CommentEvent)
async def on_comment(event: CommentEvent) -> None:
    # IDE will autocomplete event properties
    print(event.comment)

@client.on(GiftEvent)
async def on_gift(event: GiftEvent) -> None:
    print(event.gift.name)

Error Handling

Try-Except in Handlers

Always wrap your handler logic in try-except to prevent one error from breaking all event processing:

@client.on(CommentEvent)
async def on_comment(event: CommentEvent):
    try:
        # Your logic here
        await process_comment(event)
    except Exception as e:
        print(f"Error processing comment: {e}")
        # Optionally log the error
        logger.exception("Comment processing failed")

Global Error Handler

@client.on(CommentEvent)
async def on_comment(event: CommentEvent):
    await safe_handle_event(event, process_comment)

async def safe_handle_event(event, handler):
    """Wrapper for safe event handling"""
    try:
        await handler(event)
    except Exception as e:
        logger.error(f"Error in {handler.__name__}: {e}")

Common Patterns

Command Handler Pattern

COMMANDS = {
    "!help": show_help,
    "!info": show_info,
    "!stats": show_stats,
}

@client.on(CommentEvent)
async def on_comment(event: CommentEvent):
    message = event.comment.strip()
    
    if message.startswith("!"):
        command = message.split()[0].lower()
        if command in COMMANDS:
            await COMMANDS[command](event)

Gift Tracking Pattern

gift_counts = {}

@client.on(GiftEvent)
async def on_gift(event: GiftEvent):
    # Only count when gift streak ends
    if event.gift.streakable and event.streaking:
        return  # Still streaming
    
    gift_name = event.gift.name
    count = event.repeat_count
    
    gift_counts[gift_name] = gift_counts.get(gift_name, 0) + count
    print(f"Total {gift_name}: {gift_counts[gift_name]}")

State Management Pattern

class StreamStats:
    def __init__(self):
        self.comment_count = 0
        self.like_count = 0
        self.join_count = 0
        self.total_gifts = 0.0

stats = StreamStats()

@client.on(CommentEvent)
async def on_comment(event: CommentEvent):
    stats.comment_count += 1

@client.on(LikeEvent)
async def on_like(event: LikeEvent):
    stats.like_count += 1

@client.on(JoinEvent)
async def on_join(event: JoinEvent):
    stats.join_count += 1

@client.on(GiftEvent)
async def on_gift(event: GiftEvent):
    if event.value:
        stats.total_gifts += event.value

Checking for Listeners

You can check if a client is listening to a specific event:

from TikTokLive.events import CommentEvent, GiftEvent

if client.has_listener(CommentEvent):
    print("Client is listening to comments")

if not client.has_listener(GiftEvent):
    print("No gift listener registered")
    client.add_listener(GiftEvent, handle_gift)

Best Practices

Use Type Hints

Always annotate event parameters for better IDE support and code clarity.

Handle Errors

Wrap handler logic in try-except to prevent one error from breaking all events.

Keep Handlers Fast

Event handlers should be quick. For slow operations, use background tasks.

Use Async Properly

Always use async def and await for I/O operations in handlers.

Background Tasks

For long-running operations, create background tasks:

import asyncio

@client.on(GiftEvent)
async def on_gift(event: GiftEvent):
    # Quick acknowledgment
    print(f"Gift received: {event.gift.name}")
    
    # Long operation in background
    asyncio.create_task(process_gift_slowly(event))

async def process_gift_slowly(event: GiftEvent):
    """This runs in the background"""
    await asyncio.sleep(5)  # Simulate slow operation
    await save_to_database(event)

Decorator vs add_listener

Feature	`@client.on()`	`client.add_listener()`
Syntax	Cleaner, more Pythonic	More explicit
Use Case	Static event handlers	Dynamic event handlers
When to Use	Most common scenarios	Runtime configuration
Returns	Handler decorator	Handler object

When to use `@client.on()`

✅ Static event handlers defined at module level
✅ Clear, readable code with decorator syntax
✅ Most common use cases

When to use `add_listener()`

✅ Dynamic event handler registration
✅ Conditional handler setup based on config
✅ Programmatic event management
✅ Plugin/extension systems

Complete Example

from TikTokLive import TikTokLiveClient
from TikTokLive.events import (
    ConnectEvent,
    DisconnectEvent,
    CommentEvent,
    GiftEvent,
    JoinEvent,
    LiveEndEvent,
)
import asyncio

client = TikTokLiveClient(unique_id="@username")

# Connection lifecycle
@client.on(ConnectEvent)
async def on_connect(event: ConnectEvent):
    print(f"✅ Connected to @{event.unique_id} (Room {event.room_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")

# User interactions
@client.on(CommentEvent)
async def on_comment(event: CommentEvent):
    try:
        username = event.user_info.nickname
        message = event.comment
        print(f"💬 {username}: {message}")
    except Exception as e:
        print(f"Error handling comment: {e}")

@client.on(GiftEvent)
async def on_gift(event: GiftEvent):
    try:
        # Only log when gift streak ends
        if event.gift.streakable and event.streaking:
            return
        
        sender = event.from_user.nickname
        gift = event.gift.name
        count = event.repeat_count
        value = event.value
        
        print(f"🎁 {sender} sent {count}x {gift} (${value:.2f})")
    except Exception as e:
        print(f"Error handling gift: {e}")

@client.on(JoinEvent)
async def on_join(event: JoinEvent):
    print(f"👋 {event.user.nickname} joined")

if __name__ == "__main__":
    # Run the client
    client.run()

Get Started

Core Concepts

Guides

Advanced

Overview

Using Decorators

Basic Decorator Usage

Multiple Event Handlers

Decorator Method Signature

Using add_listener()

Basic add_listener Usage

Dynamic Event Registration

add_listener Method Signature

Event Handler Functions

Async Handlers

Handler Parameters

Type Hints

Error Handling

Try-Except in Handlers

Global Error Handler

Common Patterns

Command Handler Pattern

Gift Tracking Pattern

State Management Pattern

Checking for Listeners

Best Practices

Use Type Hints

Handle Errors

Keep Handlers Fast

Use Async Properly

Background Tasks

Decorator vs add_listener

When to use `@client.on()`

When to use `add_listener()`

Complete Example

Next Steps

Events Overview

Connection Lifecycle

Build docs developers (and LLMs) love

Get Started

Core Concepts

Guides

Advanced

​Overview

​Using Decorators

​Basic Decorator Usage

​Multiple Event Handlers

​Decorator Method Signature

​Using add_listener()

​Basic add_listener Usage

​Dynamic Event Registration

​add_listener Method Signature

​Event Handler Functions

​Async Handlers

​Handler Parameters

​Type Hints

​Error Handling

​Try-Except in Handlers

​Global Error Handler

​Common Patterns

​Command Handler Pattern

​Gift Tracking Pattern

​State Management Pattern

​Checking for Listeners

​Best Practices

Use Type Hints

Handle Errors

Keep Handlers Fast

Use Async Properly

​Background Tasks

​Decorator vs add_listener

​When to use @client.on()

​When to use add_listener()

​Complete Example

​Next Steps

Events Overview

Connection Lifecycle

Build docs developers (and LLMs) love

Overview

Using Decorators

Basic Decorator Usage

Multiple Event Handlers

Decorator Method Signature

Using add_listener()

Basic add_listener Usage

Dynamic Event Registration

add_listener Method Signature

Event Handler Functions

Async Handlers

Handler Parameters

Type Hints

Error Handling

Try-Except in Handlers

Global Error Handler

Common Patterns

Command Handler Pattern

Gift Tracking Pattern

State Management Pattern

Checking for Listeners

Best Practices

Background Tasks

Decorator vs add_listener

When to use `@client.on()`

When to use `add_listener()`

Complete Example

Next Steps