TikTokLive.client.ws package

Submodules

TikTokLive.client.ws.ws_client module

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

Bases: object

Websocket client responsible for connections to TikTok

build_connection_args(uri: str, headers: Dict[str, str]) → Dict[str, Any]

Build the websockets library connection arguments dictionary

Parameters:

  • uri – URI to connect to TikTok

  • headers – Headers to send to TikTok on connecting

Returns:

Connection dictionary

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.

async connect(uri: str, headers: Dict[str, str]) → AsyncIterator[WebcastResponseMessage]

Connect to the Webcast websocket server & handle cancellation

Parameters:

  • uri

  • headers

Returns:

async connect_loop(uri: str, headers: Dict[str, str]) → AsyncIterator[WebcastResponseMessage]

Connect to the Webcast server & iterate over response messages.

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.

Parameters:

  • uri – URI to connect to

  • headers – Headers used for the connection

Returns:

Yields WebcastResponseMessage

property connected: bool

Check if the websocket is currently connected

Returns:

Connection status

async disconnect() → None

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

async ping_loop() → None

Send a ping every 10 seconds to keep the connection alive

async process_recv(data: bytes) → List[WebcastResponseMessage]

Handle push frames received as websocket data

Parameters:

data – Protobuf bytestream

Returns:

List of contained messages for handling

async send_ack(log_id: int, internal_ext: str) → None

Acknowledge incoming messages from TikTok

Parameters:

  • log_id – ID for the acknowledgement

  • internal_ext – [unknown] Outbound data

Returns:

None

async send_stupid_ping() → None

Send a stupid ping with arbitrary data we found in the WS TikTok client sends this every 10 seconds from testing

TikTokLive.client.ws.ws_connect module

class TikTokLive.client.ws.ws_connect.WebcastConnect(uri: str, *, create_protocol: Callable[[...], WebSocketClientProtocol] | None = None, logger: Logger | LoggerAdapter | None = None, compression: str | None = 'deflate', origin: Origin | None = None, extensions: Sequence[ClientExtensionFactory] | None = None, subprotocols: Sequence[Subprotocol] | None = None, extra_headers: Headers | Mapping[str, str] | Iterable[Tuple[str, str]] | SupportsKeysAndGetItem | None = None, user_agent_header: str | None = 'Python/3.12 websockets/13.1', open_timeout: float | None = 10, ping_interval: float | None = 20, ping_timeout: float | None = 20, close_timeout: float | None = None, max_size: int | None = 1048576, max_queue: int | None = 32, read_limit: int = 65536, write_limit: int = 65536, **kwargs: Any)

Bases: Connect

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

Bases: ProxyConnect

Module contents