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_sockets.sockets_factory import connector_factory
from chumicro_timing import ticks_ms

client = WebSocketClient(transport_factory=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.25.0.tar.gz (74.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.25.0.tar.gz.

File metadata

File hashes

Hashes for chumicro_websockets_experimental-0.25.0.tar.gz
Algorithm Hash digest
SHA256 7ef271574ee654deedefc21f185b7b8d56acae927063077150d1c8834d4ea5a7
MD5 4ac2905b59bc11d2953ec51c5aaf8298
BLAKE2b-256 509ed3b47309204328d8a1eb6de05c0c7329f969caca7208b00f91322fa1d58a

See more details on using hashes here.

Provenance

The following attestation bundles were made for chumicro_websockets_experimental-0.25.0.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.25.0-py3-none-any.whl.

File metadata

File hashes

Hashes for chumicro_websockets_experimental-0.25.0-py3-none-any.whl
Algorithm Hash digest
SHA256 36e596e865212887d0bfce03f5208bce9647416a1fbc128d8969b66228516c4e
MD5 758af6b2b1faf5bfa8c055831e39467a
BLAKE2b-256 b469a642f8bb5a64a8fec98d90e564d2ef05c34a4b9ce0c8ebb59aee00ff6d15

See more details on using hashes here.

Provenance

The following attestation bundles were made for chumicro_websockets_experimental-0.25.0-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