WebSocket Client

import websocket

# High-level event-driven interface
from websocket import WebSocketApp, setReconnect

# Low-level connection control
from websocket import WebSocket, create_connection

# Frame protocol and status codes
from websocket import ABNF, STATUS_NORMAL, STATUS_GOING_AWAY, STATUS_PROTOCOL_ERROR

# Exception handling
from websocket import WebSocketException, WebSocketConnectionClosedException

# Debugging and logging
from websocket import enableTrace

# Socket configuration
from websocket import setdefaulttimeout, getdefaulttimeout

from websocket import create_connection

# Connect and exchange messages
ws = create_connection("ws://echo.websocket.events/")
print(ws.recv())  # Server greeting
ws.send("Hello, World")
result = ws.recv()
print(result)  # "Hello, World"
ws.close()

import websocket
import rel

def on_message(ws, message):
    print(f"Received: {message}")

def on_error(ws, error):
    print(f"Error: {error}")

def on_close(ws, close_status_code, close_msg):
    print("Connection closed")

def on_open(ws):
    print("Connection opened")

# Create WebSocket application with callbacks
ws = websocket.WebSocketApp("wss://api.gemini.com/v1/marketdata/BTCUSD",
                          on_open=on_open,
                          on_message=on_message,
                          on_error=on_error,
                          on_close=on_close)

# Run with automatic reconnection
ws.run_forever(dispatcher=rel, reconnect=5)
rel.signal(2, rel.abort)
rel.dispatch()

class WebSocketApp:
    def __init__(
        self,
        url: str,
        header: Union[list, dict, Callable, None] = None,
        on_open: Optional[Callable[[WebSocket], None]] = None,
        on_reconnect: Optional[Callable[[WebSocket], None]] = None,
        on_message: Optional[Callable[[WebSocket, Any], None]] = None,
        on_error: Optional[Callable[[WebSocket, Any], None]] = None,
        on_close: Optional[Callable[[WebSocket, Any, Any], None]] = None,
        on_ping: Optional[Callable] = None,
        on_pong: Optional[Callable] = None,
        on_cont_message: Optional[Callable] = None,
        keep_running: bool = True,
        get_mask_key: Optional[Callable] = None,
        cookie: Optional[str] = None,
        subprotocols: Optional[list] = None,
        on_data: Optional[Callable] = None,
        socket: Optional[socket.socket] = None,
    ) -> None: ...

    def run_forever(
        self,
        sockopt: tuple = None,
        sslopt: dict = None,
        ping_interval: Union[float, int] = 0,
        ping_timeout: Union[float, int, None] = None,
        ping_payload: str = "",
        http_proxy_host: str = None,
        http_proxy_port: Union[int, str] = None,
        http_no_proxy: list = None,
        http_proxy_auth: tuple = None,
        http_proxy_timeout: Optional[float] = None,
        skip_utf8_validation: bool = False,
        host: str = None,
        origin: str = None,
        dispatcher=None,
        suppress_origin: bool = False,
        proxy_type: str = None,
        reconnect: int = None,
    ) -> bool: ...

    def send(self, data: Union[bytes, str], opcode: int = ABNF.OPCODE_TEXT) -> None: ...
    def send_text(self, text_data: str) -> None: ...
    def send_bytes(self, data: Union[bytes, bytearray]) -> None: ...
    def close(self, **kwargs) -> None: ...

def setReconnect(reconnectInterval: int) -> None: ...

class WebSocket:
    def __init__(
        self,
        get_mask_key=None,
        sockopt=None,
        sslopt=None,
        fire_cont_frame: bool = False,
        enable_multithread: bool = True,
        skip_utf8_validation: bool = False,
        **_,
    ): ...

    def connect(self, url, **options): ...
    def send(self, payload: Union[bytes, str], opcode: int = ABNF.OPCODE_TEXT) -> int: ...
    def send_text(self, text_data: str) -> int: ...
    def send_bytes(self, data: Union[bytes, bytearray]) -> int: ...
    def send_frame(self, frame) -> int: ...
    def recv(self) -> Union[str, bytes]: ...
    def recv_data(self, control_frame: bool = False) -> tuple: ...
    def recv_data_frame(self, control_frame: bool = False) -> tuple: ...
    def ping(self, payload: Union[str, bytes] = ""): ...
    def pong(self, payload: Union[str, bytes] = ""): ...
    def close(self, status: int = STATUS_NORMAL, reason: bytes = b"", timeout: int = 3): ...

def create_connection(url: str, timeout=None, class_=WebSocket, **options) -> WebSocket: ...

class ABNF:
    OPCODE_CONT = 0x0
    OPCODE_TEXT = 0x1
    OPCODE_BINARY = 0x2
    OPCODE_CLOSE = 0x8
    OPCODE_PING = 0x9
    OPCODE_PONG = 0xa

    def __init__(
        self,
        fin: int = 0,
        rsv1: int = 0,
        rsv2: int = 0,
        rsv3: int = 0,
        opcode: int = OPCODE_TEXT,
        mask_value: int = 1,
        data: Union[str, bytes, None] = "",
    ) -> None: ...

    def validate(self, skip_utf8_validation: bool = False) -> None: ...
    @staticmethod
    def create_frame(data: Union[bytes, str], opcode: int, fin: int = 1) -> "ABNF": ...
    def format(self) -> bytes: ...
    @staticmethod
    def mask(mask_key: Union[str, bytes], data: Union[str, bytes]) -> bytes: ...

# Status codes
STATUS_NORMAL = 1000
STATUS_GOING_AWAY = 1001
STATUS_PROTOCOL_ERROR = 1002
STATUS_UNSUPPORTED_DATA_TYPE = 1003
STATUS_INVALID_PAYLOAD = 1007
STATUS_POLICY_VIOLATION = 1008
STATUS_MESSAGE_TOO_BIG = 1009
STATUS_INVALID_EXTENSION = 1010
STATUS_UNEXPECTED_CONDITION = 1011

class WebSocketException(Exception): ...
class WebSocketProtocolException(WebSocketException): ...
class WebSocketPayloadException(WebSocketException): ...
class WebSocketConnectionClosedException(WebSocketException): ...
class WebSocketTimeoutException(WebSocketException): ...
class WebSocketProxyException(WebSocketException): ...
class WebSocketBadStatusException(WebSocketException):
    def __init__(
        self,
        message: str,
        status_code: int,
        status_message=None,
        resp_headers=None,
        resp_body=None,
    ): ...
class WebSocketAddressException(WebSocketException): ...

def enableTrace(
    traceable: bool,
    handler: logging.StreamHandler = logging.StreamHandler(),
    level: str = "DEBUG",
) -> None: ...

def dump(title: str, message: str) -> None: ...
def error(msg: str) -> None: ...
def warning(msg: str) -> None: ...
def debug(msg: str) -> None: ...
def info(msg: str) -> None: ...
def trace(msg: str) -> None: ...
def isEnabledForError() -> bool: ...
def isEnabledForDebug() -> bool: ...
def isEnabledForTrace() -> bool: ...

DEFAULT_SOCKET_OPTION: list

class sock_opt:
    def __init__(self, sockopt: list, sslopt: dict) -> None: ...

def setdefaulttimeout(timeout: Union[int, float, None]) -> None: ...
def getdefaulttimeout() -> Union[int, float, None]: ...
def recv(sock: socket.socket, bufsize: int) -> bytes: ...
def recv_line(sock: socket.socket) -> bytes: ...
def send(sock: socket.socket, data: Union[bytes, str]) -> int: ...

# Install websocket-client to get wsdump command
pip install websocket-client

# Basic usage - connect to WebSocket URL
wsdump ws://echo.websocket.events/

# With proxy support
wsdump --proxy http://127.0.0.1:8080 ws://example.com/ws

# Enable verbose output (show opcodes)
wsdump -v ws://echo.websocket.events/

# Enable full debug tracing
wsdump -vv ws://echo.websocket.events/

# Send initial text message
wsdump --text "Hello Server" ws://echo.websocket.events/

# Ignore SSL certificate errors
wsdump --nocert wss://self-signed.badssl.com/

# Set custom headers
wsdump --headers "Authorization: Bearer token,X-Custom: value" ws://api.example.com/ws