TikTokLive.client.ws package

Submodules

TikTokLive.client.ws.ws_client module

class TikTokLive.client.ws.ws_client.WebcastWSClient(ws_kwargs: dict | None = None, ws_proxy: Proxy | AsyncioProxy | None = None)

Bases: object

Websocket client responsible for connections to TikTok

DEFAULT_PING_INTERVAL: float = 1.0
PING_MESSAGE: bytes = b'2\x02pb:\x02hb'
async connect(room_id: int, cookies: Cookies, user_agent: str, initial_webcast_response: WebcastResponse, process_connect_events: bool = True, compress_ws_events: bool = True) → AsyncIterator[WebcastResponse]

Connect to the Webcast server & iterate over response messages.

— Message 1 —

The iterator exits normally when the connection is closed with close code 1000 (OK) or 1001 (going away) or without a close code. It raises a ConnectionClosedError when the connection is closed with any other code.

— Message 2 —

DEVELOP SANITY NOTE:

When ping_timeout is set (i.e. not None), the client waits for a pong for N seconds. TikTok DO NOT SEND pongs back. Unfortunately the websockets client after N seconds assumes the server is dead. It then throws the following infamous exception:

websockets.exceptions.ConnectionClosedError: sent 1011 (unexpected error) keepalive ping timeout; no close frame received

If you set ping_timeout to None, it doesn’t wait for a pong. Perfect, since TikTok don’t send them.

— Parameters –

Parameters:

  • initial_webcast_response – The Initial WebcastResponse from the sign server - NOT a PushFrame

  • room_id – The room ID to connect to

  • user_agent – The user agent to pass to the WebSocket connection

  • cookies – The cookies to pass to the WebSocket connection

  • process_connect_events – Whether to process the initial events sent in the first fetch

  • compress_ws_events – Whether to ask TikTok to gzip the WebSocket events

Returns:

Yields WebcastResponseMessage, the messages within WebcastResponse.messages

property connected: bool

Check if the WebSocket is open

Returns:

WebSocket status

async disconnect() → None

Request to stop the websocket connection & wait :return: None

restart_ping_loop() → None

Restart the WebSocket ping loop

async send(message: bytes | Message) → None

Send a message to the WebSocket

Parameters:

message – Message to send to the WebSocket connection

async send_ack(webcast_response: WebcastResponse, webcast_push_frame: WebcastPushFrame) → None

Acknowledge the receipt of a WebcastResponse message from TikTok, if necessary

Parameters:

  • webcast_response – The WebcastResponse to acknowledge

  • webcast_push_frame – The WebcastPushFrame containing the WebcastResponse

Returns:

None

property ws: WebSocketClientProtocol | None

Get the current WebSocketClientProtocol

Returns:

WebSocketClientProtocol

TikTokLive.client.ws.ws_connect module

class TikTokLive.client.ws.ws_connect.WebcastConnect(initial_webcast_response: WebcastResponse, logger: Logger, base_uri_params: Dict[str, Any], base_uri_append_str: str, uri: str | None = None, **kwargs)

Bases: Connect

property ws: WebSocketClientProtocol | None

Get the current WebSocketClientProtocol

TikTokLive.client.ws.ws_connect.WebcastProxy

Type hint for a WebcastIterator, which yields a tuple of WebcastPushFrame and WebcastResponse. WebcastPushFrame is Optional because the first yielded item is from the initial response which is from /im/fetch (from the sign server), so it is not encapsulated by a WebcastPushFrame.

alias of Proxy | AsyncioProxy

class TikTokLive.client.ws.ws_connect.WebcastProxyConnect(proxy: Proxy | AsyncioProxy | None, **kwargs)

Bases: ProxyConnect, WebcastConnect

Add Proxy support to the WebcastConnect class

TikTokLive.client.ws.ws_utils module

TikTokLive.client.ws.ws_utils.build_webcast_uri(initial_webcast_response: WebcastResponse, base_uri_params: dict, base_uri_append_str: str) → str

Build a webcast URI from a base URI and parameters. This method will format the base URI with the parameters and return the formatted URI.

Parameters:

  • initial_webcast_response – The initial Webcast response

  • base_uri_params – Parameters to format the URI with

  • base_uri_append_str – String to append to the base URI

Returns:

str The formatted URI

TikTokLive.client.ws.ws_utils.extract_webcast_push_frame(data: bytes, logger: ~logging.Logger = <Logger TikTokLive (WARNING)>) → WebcastPushFrame

Extract a WebcastPushFrame from a raw byte payload. This method will parse the payload and return a WebcastPushFrame object. This method is useful for extracting push frames from a WebSocket connection.

Parameters:

  • data – Raw byte payload to extract from

  • logger – Logger to use for logging

Returns:

WebcastPushFrame The extracted push frame

TikTokLive.client.ws.ws_utils.extract_webcast_response_message(push_frame: ~TikTokLive.proto.tiktok_proto.WebcastPushFrame, logger: ~logging.Logger = <Logger TikTokLive (WARNING)>) → WebcastResponse

Extract the WebcastResponse from a push frame. If compression is enabled on the WebSocket, then messages will come gzipped. This method will decompress the payload if necessary. The gzip format allows for less bandwidth usage, at the cost of a slight CPU increase for message decompression.

Parameters:

  • push_frame – Push frame to extract from

  • logger – Logger to use for logging

Returns:

WebcastResponse The extracted response

Module contents