WSX Protocol

WebSocket extension protocol for genro-asgi.

Protocol

WSX protocol utilities.

WSX (WebSocket eXtended) is a protocol that brings HTTP-like semantics to WebSocket and NATS messaging. Messages are prefixed with WSX:// followed by JSON containing id, method, path, headers, cookies, query, data.

Format:

WSX://{“id”:”uuid”,”method”:”POST”,”path”:”/users”,”headers”:{},”data”:{}}

This module provides: - WSX_PREFIX: Protocol prefix constant - is_wsx_message(): Check if data is WSX format - parse_wsx_message(): Parse WSX message to dict - build_wsx_message(): Build WSX message from components - build_wsx_response(): Build WSX response message

Request handling uses MsgRequest from request.py which calls these functions.

genro_asgi.wsx.protocol.build_wsx_message(*, id, method, path='/', headers=None, cookies=None, query=None, data=None, tytx=False)[source]

Build a WSX request message string.

Parameters:

  • id (str) – Correlation ID (required)

  • method (str) – HTTP method (required)

  • path (str) – Request path (default: “/”)

  • headers (dict[str, str] | None) – Headers dict

  • cookies (dict[str, str] | None) – Cookies dict

  • query (dict[str, Any] | None) – Query parameters

  • data (Any) – Payload data

  • tytx (bool) – Whether to use TYTX serialization

Returns:

// prefixed message string

Return type:

str

genro_asgi.wsx.protocol.build_wsx_response(*, id, status=200, headers=None, cookies=None, data=None)[source]

Build a WSX response message string.

Parameters:

  • id (str) – Correlation ID from request (required)

  • status (int) – HTTP status code (default: 200)

  • headers (dict[str, str] | None) – Response headers dict

  • cookies (dict[str, Any] | None) – Response cookies dict

  • data (Any) – Response payload

Returns:

// prefixed response message string

Return type:

str

genro_asgi.wsx.protocol.is_wsx_message(data)[source]

Check if data is a WSX protocol message.

Parameters:

data (str | bytes) – String or bytes to check

Returns:

// prefix

Return type:

bool

genro_asgi.wsx.protocol.parse_wsx_message(data)[source]

Parse a WSX message into a dictionary.

Parameters:

data (str | bytes) – WSX message (with or without prefix)

Return type:

dict[str, Any]

Returns:

Parsed message dict with id, method, path, headers, etc.

Raises: