Skip to main content

Non-blocking WebSocket (RFC 6455) client + server for CircuitPython, MicroPython, and CPython

Project description

chumicro-websockets

Non-blocking WebSocket (RFC 6455) client and server, plain and TLS.

RFC 6455 framing and masking, opening + closing handshakes, text + binary + ping/pong, and oversized-message guards. Plays alongside chumicro-http-server for combined HTTP + WS / HTTPS + WSS deployments. Built on chumicro-sockets and chumicro-timing.


Part of the ChuMicro family — small, focused Python libraries for microcontrollers and laptops. Browse all libraries.

Install

# CircuitPython (after `circup bundle-add ChuMicro/ChuMicro-Bundle-Experimental`)
circup install chumicro_websockets

# MicroPython
mpremote mip install github:ChuMicro/ChuMicro-Bundle-Experimental/chumicro_websockets

# CPython
pip install chumicro-websockets-experimental

For bundle setup, pre-compiled .mpy bundles, the experimental channel, and details on PyPI naming, see the chumicro INSTALL guide.

Quick example

Client:

from chumicro_websockets import WebSocketClient, WebSocketState
from chumicro_websockets.sockets_factory import chumicro_sockets_connector_factory
from chumicro_timing import ticks_ms

client = WebSocketClient(transport_factory=chumicro_sockets_connector_factory())
client.on_text = lambda text: print(f"got: {text}")
client.connect("ws://api.example.com/stream")

while client.state != WebSocketState.CLOSED:
    if client.check(ticks_ms()):
        client.handle(ticks_ms())

Server:

from chumicro_websockets import WebSocketServer
from chumicro_sockets import listener
from chumicro_timing import ticks_ms

def on_connection(connection):
    connection.on_text = lambda text: connection.send_text(f"echo: {text}")

server = WebSocketServer(
    listener=listener("0.0.0.0", 8765),
    on_connection=on_connection,
)

while True:
    if server.check(ticks_ms()):
        server.handle(ticks_ms())

What's included

Component Purpose
WebSocketClient Non-blocking RFC 6455 client; runner-shaped check(now_ms) / handle(now_ms).
WebSocketServer + Connection Standalone WebSocket server owning a listening socket; one Connection per accepted client.
WhenOversized Policy for inbound messages above max_message_bytes: DROP_SILENT, DROP_WITH_EVENT, DISCONNECT.
WebSocketState Lifecycle constants (CONNECTING / OPEN / CLOSING / CLOSED).
parse_ws_url(url) Split ws:// / wss:// URLs into (scheme, host, port, path).
make_websocket_key() / derive_accept_key(key) RFC 6455 §4.2.2 nonce + accept-token derivation.
WebSocketError + subclasses Exception hierarchy: protocol, handshake, URL, timeout, backpressure, state.
OPCODE_* / CLOSE_* constants RFC 6455 opcode and close-code values.

Wire-format primitives (FrameParser, encode_frame, the handshake parsers and encoders, the close-payload codec, validate_text_payload, the DEFAULT_* knob constants) live in chumicro_websockets._wire for advanced users and tests; the public top-level surface stays small to keep flash + RAM lean on the device.

Where this fits

Depends on chumicro-sockets and chumicro-timing. Pairs with chumicro-http-server for combined HTTP + WS / HTTPS + WSS deployments.

Platform support

Works on CPython, MicroPython, and CircuitPython.

Examples

Example What it shows
client.py WebSocket client on real CP/MP hardware — brings wifi up via the bundled helpers, connects to a configured echo server, prints every inbound message while a counter ticks alongside.
server.py WebSocket echo server on real CP/MP hardware — accepts inbound connections on the configured host/port and echoes every frame back.

Contributing

Working on chumicro-websockets itself? Clone the mono-repo if you haven't already — the rest of the workflow assumes you're inside that workspace.

pip install -e .[test]
pytest tests/                  # host-side tests
pytest functional_tests/       # on-device tests (needs a board registered in devices.yml)

Register a board before running functional tests: chumicro-workspace add-device <id> --address <port>.

Docs

📖 Stable docs · Experimental docs

Find this library

License

MIT

Project details


Download files

Download the file for your platform. If you're not sure which to choose, learn more about installing packages.

Source Distribution

chumicro_websockets_experimental-0.24.2.tar.gz (75.7 kB view details)

Uploaded Source

Built Distribution

If you're not sure about the file name format, learn more about wheel file names.

File details

Details for the file chumicro_websockets_experimental-0.24.2.tar.gz.

File metadata

File hashes

Hashes for chumicro_websockets_experimental-0.24.2.tar.gz
Algorithm Hash digest
SHA256 e5d5d24a6746d61e3b4e1b5135b8accf44c665e2114eb2c967eeba4177b835e9
MD5 97dfb08945050fc7e93f88bae1728251
BLAKE2b-256 9a3ad2552aaee04fcde6e2803b79178d8639fec39ca83e3769de2838beeb1ef0

See more details on using hashes here.

Provenance

The following attestation bundles were made for chumicro_websockets_experimental-0.24.2.tar.gz:

Publisher: release.yml on ChuMicro/ChuMicro

Attestations: Values shown here reflect the state when the release was signed and may no longer be current.

File details

Details for the file chumicro_websockets_experimental-0.24.2-py3-none-any.whl.

File metadata

File hashes

Hashes for chumicro_websockets_experimental-0.24.2-py3-none-any.whl
Algorithm Hash digest
SHA256 4fc6a073c2c59b5df475af25088842e70574c82e4370f4393aaa4a5df5251a76
MD5 05ad18a1028787f7bc9cc1058f0f8310
BLAKE2b-256 678e52114fd5d8222a562653cfb869e65570442ce2b58bc3952947e21048fe61

See more details on using hashes here.

Provenance

The following attestation bundles were made for chumicro_websockets_experimental-0.24.2-py3-none-any.whl:

Publisher: release.yml on ChuMicro/ChuMicro

Attestations: Values shown here reflect the state when the release was signed and may no longer be current.

Supported by

AWS Cloud computing and Security Sponsor Datadog Monitoring Depot Continuous Integration Fastly CDN Google Download Analytics Pingdom Monitoring Sentry Error logging StatusPage Status page