WebcastWSClient

Overview

The WebcastWSClient class is responsible for managing WebSocket connections to TikTok LIVE streams. It handles connection management, message sending/receiving, acknowledgments, and heartbeat pings to keep the connection alive.

Class Definition

from TikTokLive.client.ws.ws_client import WebcastWSClient

Location

TikTokLive/client/ws/ws_client.py:17

Constructor

WebcastWSClient(
    ws_kwargs: Optional[dict] = None,
    ws_proxy: Optional[WebcastProxy] = None
)

Parameters

ws_kwargs

dict

Optional dictionary of keyword arguments to override WebSocket connection settings. These kwargs are passed to the underlying websockets library.Common Options:

subprotocols: List of WebSocket subprotocols (default: ["echo-protocol"])
extra_headers: Dictionary of additional HTTP headers to send during handshake
base_uri_params: Additional query parameters for the WebSocket URI
base_uri_append_str: String to append to the WebSocket URI
proxy_conn_timeout: Proxy connection timeout in seconds (default: 10.0 when using proxy)
ping_interval: Custom ping interval (handled internally by TikTok)
ping_timeout: Custom ping timeout (set to None internally as TikTok doesn’t send pongs)

ws_proxy

WebcastProxy

Optional proxy for the WebSocket connection. Can be either:

httpx.Proxy - Automatically converted to websockets proxy format
websockets_proxy.Proxy - Used directly

The proxy is used for all WebSocket traffic to TikTok servers.

Properties

`ws`

@property
def ws(self) -> Optional[WebSocketClientProtocol]:

Returns the current WebSocket connection protocol, or None if not connected. Returns: Optional[WebSocketClientProtocol] Location: ws_client.py:42

`connected`

@property
def connected(self) -> bool:

Checks if the WebSocket connection is currently open and active. Returns: bool - True if connected, False otherwise Location: ws_client.py:58

Methods

`connect()`

async def connect(
    room_id: int,
    cookies: httpx.Cookies,
    user_agent: str,
    initial_webcast_response: ProtoMessageFetchResult,
    process_connect_events: bool = True,
    compress_ws_events: bool = True
) -> AsyncIterator[ProtoMessageFetchResult]

Connect to the TikTok Webcast server and iterate over response messages.

Parameters

room_id

int

required

The TikTok LIVE room ID to connect to

httpx.Cookies

required

HTTP cookies to include in the WebSocket handshake (includes session authentication)

user_agent

str

required

User-Agent string to send in the WebSocket handshake headers

initial_webcast_response

ProtoMessageFetchResult

required

The initial response from the sign server containing WebSocket connection details

process_connect_events

bool

default:"True"

Whether to process and yield events from the initial response

compress_ws_events

bool

default:"True"

Whether to request gzip compression for WebSocket events (reduces bandwidth usage)

Returns

AsyncIterator[ProtoMessageFetchResult] - Async iterator yielding message results from the stream

Important Notes

Ping Timeout Behavior: TikTok does not send WebSocket pong responses. The client sets ping_timeout=None internally to prevent connection timeouts. Do not override this setting.

The iterator exits normally when the connection closes with code 1000 (OK) or 1001 (going away). Other close codes raise ConnectionClosedError.

Location: ws_client.py:167

`send()`

async def send(message: Union[bytes, Message]) -> None

Send a message to the WebSocket server.

Parameters

message

Union[bytes, Message]

required

Message to send. Can be raw bytes or a protobuf Message object (will be serialized automatically)

Location: ws_client.py:69

`send_ack()`

async def send_ack(
    webcast_response: ProtoMessageFetchResult,
    webcast_push_frame: WebcastPushFrame
) -> None

Acknowledge receipt of a message from TikTok. This is required for certain message types.

Parameters

webcast_response

ProtoMessageFetchResult

required

The message result to acknowledge

webcast_push_frame

WebcastPushFrame

required

The push frame containing the message (provides log_id for acknowledgment)

Location: ws_client.py:90

`disconnect()`

async def disconnect() -> None

Close the WebSocket connection gracefully. Location: ws_client.py:120

`switch_rooms()`

async def switch_rooms(room_id: int) -> None

Switch to a different LIVE room without disconnecting. Sends an im_enter_room message and restarts the ping loop.

Parameters

room_id

int

required

The new room ID to switch to

Location: ws_client.py:296

`get_ws_cookie_string()`

def get_ws_cookie_string(cookies: httpx.Cookies) -> str

Generate a cookie string for the WebSocket connection from HTTP cookies.

Parameters

httpx.Cookies

required

HTTP cookies to format for WebSocket handshake

Returns

str - Cookie string formatted for WebSocket headers

This method logs differently for authenticated (with sessionid) vs anonymous sessions, with session IDs redacted in logs for security.

Location: ws_client.py:132

Internal Methods

`restart_ping_loop()`

def restart_ping_loop(room_id: int) -> None

Restart the internal heartbeat ping loop. Called automatically when switching rooms. Location: ws_client.py:284

`_ping_loop_fn()`

async def _ping_loop_fn(room_id: int) -> None

Internal coroutine that sends heartbeat messages every 5 seconds (or custom interval from server) to keep the connection alive. Location: ws_client.py:319

Constants

`DEFAULT_PING_INTERVAL`

DEFAULT_PING_INTERVAL: float = 5.0

Default interval in seconds between heartbeat pings. Can be overridden by server via Handshake-Options header. Location: ws_client.py:20

Usage Example

from TikTokLive.client.ws.ws_client import WebcastWSClient
import httpx

# Create WebSocket client with custom settings
ws_client = WebcastWSClient(
    ws_kwargs={
        "extra_headers": {
            "Origin": "https://www.tiktok.com"
        },
        "proxy_conn_timeout": 15.0
    },
    ws_proxy=httpx.Proxy("http://proxy.example.com:8080")
)

# Connect and process messages
async for response in ws_client.connect(
    room_id=123456789,
    cookies=cookies,
    user_agent="Mozilla/5.0...",
    initial_webcast_response=initial_response,
    compress_ws_events=True
):
    # Process messages
    for message in response.messages:
        print(f"Received: {message}")

Client

Events

Web Client

WebSocket

Errors

Overview

Class Definition

Location

Constructor

Parameters

Properties

`ws`

`connected`

Methods

`connect()`

Parameters

Returns

Important Notes

`send()`

Parameters

`send_ack()`

Parameters

`disconnect()`

`switch_rooms()`

Parameters

`get_ws_cookie_string()`

Parameters

Returns

Internal Methods

`restart_ping_loop()`

`_ping_loop_fn()`

Constants

`DEFAULT_PING_INTERVAL`

Usage Example

See Also

Build docs developers (and LLMs) love

Client

Events

Web Client

WebSocket

Errors

​Overview

​Class Definition

​Location

​Constructor

​Parameters

​Properties

​ws

​connected

​Methods

​connect()

​Parameters

​Returns

​Important Notes

​send()

​Parameters

​send_ack()

​Parameters

​disconnect()

​switch_rooms()

​Parameters

​get_ws_cookie_string()

​Parameters

​Returns

​Internal Methods

​restart_ping_loop()

​_ping_loop_fn()

​Constants

​DEFAULT_PING_INTERVAL

​Usage Example

​See Also

Build docs developers (and LLMs) love

Overview

Class Definition

Location

Constructor

Parameters

Properties

`ws`

`connected`

Methods

`connect()`

Parameters

Returns

Important Notes

`send()`

Parameters

`send_ack()`

Parameters

`disconnect()`

`switch_rooms()`

Parameters

`get_ws_cookie_string()`

Parameters

Returns

Internal Methods

`restart_ping_loop()`

`_ping_loop_fn()`

Constants

`DEFAULT_PING_INTERVAL`

Usage Example

See Also