Module

protocols.ws

WebSocket protocol handler — sans-I/O wrapper around wsproto.

Translates between raw bytes and typed WebSocket events. The worker feeds bytes in via receive_data() and reads serialized bytes out via send_message() and close().

Requires thewsproto optional dependency (pip install pounce[ws]).

Architecture: The HTTP upgrade handshake (101 Switching Protocols) is handled by the H1 layer. WSProtocol only handles WebSocket framing after the upgrade is complete. wsproto 1.x server connections start in OPEN state — no separate accept step in wsproto itself.

All state is per-connection. No shared mutable state.

1Class3Functions

Classes

WSProtocol 6 ▼
WebSocket protocol handler backed by wsproto. Sans-I/O: accepts raw bytes, produces raw bytes. The…

WebSocket protocol handler backed by wsproto.

Sans-I/O: accepts raw bytes, produces raw bytes. The worker manages the actual socket I/O.

This handler manages WebSocket framing after the HTTP upgrade handshake. Usebuild_101_response()to generate the handshake response separately.

In wsproto 1.x the server connection starts in OPEN state — the HTTP handshake is not managed by wsproto.

Methods

is_closed 0 bool ▼
True if the WebSocket connection has been closed.
property
def is_closed(self) -> bool
Returns
bool
subprotocol 0 str | None ▼
The negotiated subprotocol (if any).
property
def subprotocol(self) -> str | None
Returns
str | None
receive_data 1 tuple[list[ProtocolEvent… ▼
Feed raw bytes from the socket, return WebSocket events and outbound data.
def receive_data(self, data: bytes) -> tuple[list[ProtocolEvent], bytes]
Parameters
Name Type Description
data —

Raw WebSocket frame bytes from the network.
Returns
tuple[list[ProtocolEvent], bytes] Tuple of (protocol_events, outbound_bytes). The outbound bytes include pong responses and other protocol-level frames that must be written back to the socket.
send_message 1 bytes ▼
Serialize a WebSocket message frame.
def send_message(self, data: bytes | str) -> bytes
Parameters
Name Type Description
data —

The message payload (bytes for binary, str for text).
Returns
bytes Serialized WebSocket frame bytes.
close 2 bytes ▼
Serialize a WebSocket close frame.
def close(self, code: int = 1000, reason: str = '') -> bytes
Parameters
Name Type Description
code —

Close status code (default: 1000 Normal Closure).

Default:1000
reason —

Optional human-readable close reason.

Default:''
Returns
bytes Serialized WebSocket close frame bytes.
Internal Methods 1 ▼
__init__ 1 ▼
def __init__(self, *, subprotocol: str | None = None) -> None
Parameters
Name Type Description
subprotocol — Default:None

Functions

is_wsproto_available 0 bool ▼
Check if wsproto is installed.
def is_wsproto_available() -> bool
Returns
bool
build_ws_accept_key 1 bytes ▼
Compute the Sec-WebSocket-Accept value for the handshake.
def build_ws_accept_key(ws_key: bytes) -> bytes
Parameters
Name Type Description
ws_key bytes

The Sec-WebSocket-Key from the client's request.
Returns
bytes
build_101_response 2 bytes ▼
Build the raw HTTP 101 Switching Protocols response. This is the WebSocket han…
def build_101_response(ws_key: bytes, *, subprotocol: str | None = None) -> bytes

Build the raw HTTP 101 Switching Protocols response.

This is the WebSocket handshake response sent by the server before switching to WebSocket framing. Constructed manually because the H1 protocol handler may not support 101 responses cleanly.

Parameters
Name Type Description
ws_key bytes

The Sec-WebSocket-Key from the client.
subprotocol str | None

Optional negotiated subprotocol.

Default:None
Returns
bytes