WebSocket Connection

Overview

The WebSocket connection module provides low-level connection management for TikTok LIVE streams. It includes custom implementations that handle TikTok-specific requirements like signed URLs, compression, and proxy support.

Classes

`WebcastConnect`

Custom WebSocket connection class that extends websockets.legacy.client.Connect to handle TikTok LIVE-specific requirements.

from TikTokLive.client.ws.ws_connect import WebcastConnect

Location: TikTokLive/client/ws/ws_connect.py:27

Constructor

WebcastConnect(
    initial_webcast_response: ProtoMessageFetchResult,
    logger: logging.Logger,
    base_uri_params: Dict[str, Any],
    base_uri_append_str: str,
    uri: Optional[str] = None,
    **kwargs
)

Parameters

initial_webcast_response

ProtoMessageFetchResult

required

Initial response from the sign server containing WebSocket URL and parameters

logger

logging.Logger

required

Logger instance for connection logging

base_uri_params

Dict[str, Any]

required

Base query parameters to include in the WebSocket URI

base_uri_append_str

str

required

Additional string to append to the constructed WebSocket URI

uri

str

Optional pre-built URI (bypasses automatic URI construction)

**kwargs

Any

Additional keyword arguments passed to the parent Connect class

Properties

`ws`

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

Returns the current active WebSocket protocol. Location: ws_connect.py:54

`ws_options`

@property
def ws_options(self) -> Optional[dict[str, str]]

Returns WebSocket options received from the server via the Handshake-Options header. Location: ws_connect.py:60

Async Iterator

async def __aiter__(self) -> WebcastIterator

Custom async iterator implementation that:

Disables automatic retry mechanisms (signed URLs expire after 30 seconds)
Handles TikTok’s status code 200 as an error (detected/blocked connection)
Yields (WebcastPushFrame, ProtoMessageFetchResult) tuples

Important: The first yielded item has None for the push frame since it comes from the initial /im/fetch response. Location: ws_connect.py:66

This implementation disables reconnection attempts because TikTok’s signed WebSocket URLs expire after 30 seconds and cannot be reused.

`WebcastProxyConnect`

Extends WebcastConnect to add proxy support for WebSocket connections.

from TikTokLive.client.ws.ws_connect import WebcastProxyConnect

Location: TikTokLive/client/ws/ws_connect.py:121

Constructor

WebcastProxyConnect(
    proxy: Optional[WebcastProxy],
    **kwargs
)

Parameters

proxy

WebcastProxy

Proxy to use for the WebSocket connection. Can be:

httpx.Proxy - Automatically converted to websockets_proxy.Proxy
websockets_proxy.Proxy - Used directly

**kwargs

Any

All other parameters are passed to the parent WebcastConnect class

Proxy Conversion

@classmethod
def _convert_proxy(cls, proxy: httpx.Proxy) -> websockets_proxy.Proxy

Internal method that converts httpx.Proxy objects to websockets_proxy.Proxy format, preserving authentication credentials. Location: ws_connect.py:137

Type Definitions

`WebcastProxy`

WebcastProxy: Type = Union[httpx.Proxy, websockets_proxy.Proxy]

Type hint for proxy objects that can be used with WebSocket connections. Location: ws_connect.py:17

`WebcastIterator`

WebcastIterator: Type = AsyncIterator[Tuple[Optional[WebcastPushFrame], ProtoMessageFetchResult]]

Type hint for the async iterator that yields WebSocket messages. The WebcastPushFrame is Optional because the first yielded item is from the initial response and not encapsulated in a push frame. Location: ws_connect.py:24

Utility Functions

`build_webcast_uri()`

def build_webcast_uri(
    initial_webcast_response: ProtoMessageFetchResult,
    base_uri_params: dict,
    base_uri_append_str: str
) -> str

Builds the complete WebSocket URI from the initial response and configuration.

Parameters

initial_webcast_response

ProtoMessageFetchResult

required

Initial response containing:

push_server: WebSocket base URL
cursor: Connection cursor
route_params: Server-provided routing parameters
internal_ext: Internal extension data

base_uri_params

dict

required

Additional query parameters to include (e.g., room_id, compress)

base_uri_append_str

str

required

String to append to the constructed URI

Returns

str - Complete WebSocket URI with all parameters

Raises

InitialCursorMissingError - If the cursor is missing from the response
WebsocketURLMissingError - If the push server URL or route parameters are missing

Location: TikTokLive/client/ws/ws_utils.py:14

`extract_webcast_push_frame()`

def extract_webcast_push_frame(
    data: bytes,
    logger: logging.Logger = TikTokLiveLogHandler.get_logger()
) -> WebcastPushFrame

Parses raw bytes into a WebcastPushFrame object.

Parameters

data

bytes

required

Raw byte payload from WebSocket

logger

logging.Logger

Logger for error reporting

Returns

WebcastPushFrame - Parsed push frame object Location: ws_utils.py:58

`extract_webcast_response_message()`

def extract_webcast_response_message(
    push_frame: WebcastPushFrame,
    logger: logging.Logger = TikTokLiveLogHandler.get_logger()
) -> ProtoMessageFetchResult

Extracts and decompresses (if necessary) the ProtoMessageFetchResult from a push frame.

Parameters

push_frame

WebcastPushFrame

required

Push frame to extract from

logger

logging.Logger

Logger for error reporting

Returns

ProtoMessageFetchResult - The extracted message result

Compression Handling

This function automatically handles gzip decompression when:

The push frame has a compress_type header set to "gzip"
The payload data is gzip-compressed

Gzip compression significantly reduces bandwidth usage at the cost of minimal CPU overhead for decompression.

Location: ws_utils.py:74

`extract_websocket_options()`

def extract_websocket_options(headers: dict) -> dict[str, str]

Parses WebSocket options from the Handshake-Options header.

Parameters

headers

dict

required

HTTP headers from the WebSocket handshake response

Returns

dict[str, str] - Parsed options dictionary (e.g., {"ping-interval": "5.0"}) Location: ws_utils.py:108

Connection Flow

Error Handling

Status Code 200 Error

TikTok uses HTTP status code 200 to indicate a rejected/blocked WebSocket connection:

try:
    async for frame, response in webcast_connect:
        # Process messages
        pass
except WebcastBlocked200Error as e:
    print(f"Connection blocked: {e}")
    # Error message includes Handshake-Msg header content

A status code 200 during WebSocket handshake indicates TikTok has detected and rejected the connection. Common causes:

Invalid or missing session ID cookie when required
Suspected bot/automated access
IP-based rate limiting

Missing Required Data

The connection process validates that all required data is present:

try:
    uri = build_webcast_uri(response, params, append_str)
except InitialCursorMissingError:
    print("Cursor missing from initial response")
except WebsocketURLMissingError:
    print("WebSocket URL or parameters missing")

Usage Example

from TikTokLive.client.ws.ws_connect import WebcastProxyConnect
import httpx
import logging

# Create proxy-enabled connection
connection = WebcastProxyConnect(
    initial_webcast_response=initial_response,
    logger=logging.getLogger("tiktok"),
    base_uri_params={
        "room_id": 123456789,
        "compress": "gzip"
    },
    base_uri_append_str="&aid=1988",
    proxy=httpx.Proxy("http://proxy.example.com:8080"),
    proxy_conn_timeout=15.0,
    subprotocols=["echo-protocol"]
)

# Iterate over messages
async for push_frame, response in connection:
    if push_frame is None:
        print("Initial response received")
    else:
        print(f"Push frame type: {push_frame.payload_type}")
    
    # Process messages in response
    for message in response.messages:
        print(f"Message: {message}")

Client

Events

Web Client

WebSocket

Errors

​Overview

​Classes

​WebcastConnect

​Constructor

Parameters

​Properties

ws

ws_options

​Async Iterator

​WebcastProxyConnect

​Constructor

Parameters

​Proxy Conversion

​Type Definitions

​WebcastProxy

​WebcastIterator

​Utility Functions

​build_webcast_uri()

​Parameters

​Returns

​Raises

​extract_webcast_push_frame()

​Parameters

​Returns

​extract_webcast_response_message()

​Parameters

​Returns

​Compression Handling

​extract_websocket_options()

​Parameters

​Returns

​Connection Flow

​Error Handling

​Status Code 200 Error

​Missing Required Data

​Usage Example

​See Also

Build docs developers (and LLMs) love

Overview

Classes

`WebcastConnect`

Constructor

Properties

`ws`

`ws_options`

Async Iterator

`WebcastProxyConnect`

Constructor

Proxy Conversion

Type Definitions

`WebcastProxy`

`WebcastIterator`

Utility Functions

`build_webcast_uri()`

Parameters

Returns

Raises

`extract_webcast_push_frame()`

Parameters

Returns

`extract_webcast_response_message()`

Parameters

Returns

Compression Handling

`extract_websocket_options()`

Parameters

Returns

Connection Flow

Error Handling

Status Code 200 Error

Missing Required Data

Usage Example

See Also