WebSocket Client¶

Zero-dependency RFC 6455 WebSocket client with sync + async support -- stdlib only, Python 3.10+.

Replaces: websocket-client, websockets

Overview¶

The websocket module provides synchronous and asynchronous WebSocket clients for ws:// and wss:// connections. It implements the core WebSocket protocol including text frames, ping/pong, close handshake, and client-side masking. Designed as a drop-in alternative to the websockets library for projects that need WebSocket communication without third-party dependencies.

File	Description	Dependencies
`websocket.py`	Pure Python implementation	None (stdlib only: `asyncio`, `socket`, `ssl`, `hashlib`, `struct`)

Features¶

RFC 6455 compliant -- full WebSocket protocol with upgrade handshake, frame encoding/decoding, and masking
Sync + async clients -- WebSocketClient and AsyncWebSocketClient with identical APIs
TLS support -- wss:// connections with configurable certificate verification
Ping/pong -- automatic pong response on receiving ping frames
Close handshake -- proper close frame exchange with status codes and reasons
Custom headers -- pass extra headers in the upgrade request (auth tokens, cookies, etc.)
Subprotocol negotiation -- request and detect accepted subprotocols
Timeout control -- configurable timeouts for connect, send, and receive operations
Context manager -- with / async with for automatic connect and close

How to Use in Your Project¶

Copy the single .py file into your project:

cp websocket/websocket.py your_project/

Then import directly:

from websocket import WebSocketClient, AsyncWebSocketClient

Usage Examples¶

Basic Sync Client¶

from websocket import WebSocketClient

with WebSocketClient("ws://localhost:9222/") as ws:
    ws.send("hello")
    response = ws.recv()
    print(response)

Basic Async Client¶

import asyncio
from websocket import AsyncWebSocketClient

async def main():
    async with AsyncWebSocketClient("wss://example.com/ws") as ws:
        await ws.send('{"type": "subscribe"}')
        data = await ws.recv()
        print(data)

asyncio.run(main())

JSON-RPC over WebSocket¶

import json
from websocket import WebSocketClient

with WebSocketClient("ws://localhost:9222/devtools/page/123") as ws:
    # Send a CDP command
    ws.send(json.dumps({
        "id": 1,
        "method": "Page.navigate",
        "params": {"url": "https://example.com"}
    }))
    result = json.loads(ws.recv())
    print(result)

Custom Headers¶

from websocket import WebSocketClient

headers = {
    "Authorization": "Bearer my-token",
    "X-Custom-Header": "value",
}
with WebSocketClient("ws://localhost:8080/ws", headers=headers) as ws:
    ws.send("authenticated message")
    print(ws.recv())

Receive Timeout¶

from websocket import WebSocketClient, WebSocketTimeoutError

with WebSocketClient("ws://localhost:8080/ws") as ws:
    ws.send("ping")
    try:
        response = ws.recv(timeout=5.0)
    except WebSocketTimeoutError:
        print("No response within 5 seconds")

Manual Connect / Close¶

from websocket import WebSocketClient

ws = WebSocketClient("ws://localhost:8080/ws")
ws.connect(timeout=10)

ws.send("hello")
print(ws.recv())

ws.close(code=1000, reason="done")

Secure Connection (WSS)¶

from websocket import WebSocketClient

# With certificate verification (default)
with WebSocketClient("wss://echo.websocket.org") as ws:
    ws.send("secure hello")
    print(ws.recv())

# Without certificate verification (e.g., self-signed certs)
ws = WebSocketClient("wss://localhost:9443/ws")
ws.connect(verify=False)

API Reference¶