WSX Protocol

Version: 1.1.0 Status: SOURCE OF TRUTH Last Updated: 2026-06-28

Overview

WSX (WebSocket eXtended) is a message format for RPC over WebSocket. It brings HTTP-like semantics to message-based transports.

Important: WSX is a protocol/format specification. Request handling uses MsgRequest(BaseRequest) from request.py. WebSocket connections speaking WSX are driven by WsxHandler (wsx/handler.py).

NATS is not implemented. The format is designed to be transport-agnostic, so a NATS transport is a roadmap item only; the sections marked (roadmap) below describe intended, not existing, behaviour.

Motivation

Different transports have different APIs:

HTTP: method, path, headers, cookies, query, body
WebSocket: just binary/text messages
NATS (roadmap): subject, payload bytes, reply-to

WSX defines a message format that encapsulates HTTP-like semantics:

HTTP Request  ─────┐
                   │
WebSocket RPC ─────┼──► BaseRequest ──► Handler ──► WSX response
                   │
NATS Message  ─────┘  (roadmap)

Message Format

Prefix

WSX messages start with WSX:// followed by JSON:

WSX://{"id":"...","method":"...","path":"...","headers":{},"data":{}}

WSX Request

WSX://{
    "id": "uuid-123",
    "method": "POST",
    "path": "/users/42",
    "headers": {
        "content-type": "application/json",
        "authorization": "Bearer xxx"
    },
    "cookies": {
        "session_id": "xyz-789"
    },
    "query": {
        "limit": "10::L",
        "active": "true::B"
    },
    "data": {
        "name": "Mario",
        "birth": "1990-05-15::D"
    }
}

Request Fields

Field	Type	Required	Description
`id`	string	Yes	Correlation ID
`method`	string	Yes	GET, POST, PUT, DELETE, PATCH
`path`	string	No (defaults to `/`)	Routing path (e.g., “/users/42”)
`headers`	object	No	HTTP headers as dict
`cookies`	object	No	Cookies as dict
`query`	object	No	Query parameters (TYTX supported in `::JS` mode)
`data`	any	No	Request payload (TYTX supported in `::JS` mode)
`tytx`	bool	No	When `true`, marks the message as TYTX-encoded

WSX Response

WSX://{
    "id": "uuid-123",
    "status": 200,
    "headers": {
        "content-type": "application/json"
    },
    "cookies": {
        "session_id": "new-xyz"
    },
    "data": {
        "id": 42,
        "message": "User created"
    }
}

Response Fields

Field	Type	Required	Description
`id`	string	Yes	Same correlation ID as request
`status`	int	Yes	HTTP status code (defaults to 200)
`headers`	object	No	Response headers (omitted when empty)
`cookies`	object	No	Set-Cookie equivalents (omitted when empty)
`data`	any	No	Response payload (omitted when `None`)

A response field is only present in the JSON when it has a value: build_wsx_response always emits id and status, and adds headers, cookies, data only when they are non-empty / non-None.

TYTX Integration

Values can carry TYTX type suffixes for type preservation:

"price": "99.50::N"     → Decimal("99.50")
"date": "2025-01-15::D" → date(2025, 1, 15)
"count": "42::L"        → 42 (int)
"active": "true::B"     → True (bool)

TYTX hydration happens on the request side, in request.py — MsgRequest._parse_wsx_message() is the only place that calls genro_tytx. The wsx/protocol.py module itself uses stdlib json only and performs no TYTX hydration.

MsgRequest._parse_wsx_message() selects the decoder by inspecting the raw message:

bytes → attempts from_tytx(data, transport="msgpack"); on ImportError it decodes the bytes as UTF-8 and continues as a string.
string starting with WSX:// → the prefix is stripped.
string ending with ::JS → TYTX JSON marker; parsed with from_tytx(data). On ImportError the ::JS marker is stripped and the body is parsed as plain JSON.
any other string → parsed with stdlib json.loads.

MsgRequest also detects “TYTX mode” from the message: parsed["tytx"] is True or a content-type header containing tytx.

Transport Flow

WebSocket (ASGI)

Driven by WsxHandler.__call__ and WsxHandler._dispatch_message:

1. WebSocket.accept() → assign connection_id (uuid4)
2. Authenticate the handshake (WsxHandler._authenticate) → scope["auth"], scope["_filters"]
3. Register the connection in WsxRegistry
4. Receive loop over ws.iter_text():
   a. is_wsx_message(raw)?  (checks the WSX:// prefix) — non-WSX text is ignored
   b. parse_wsx_message(raw) → dict (stdlib json)
   c. path starts with "_wsx/" → handled inline by _handle_control (never routed)
   d. otherwise → _dispatch_message as a concurrent asyncio task (bounded by a semaphore)
5. _dispatch_message:
   - request_registry.create(..., message=raw, websocket=ws) → MsgRequest
     (this is where TYTX hydration runs, in request.py)
   - demultiplex on the first path segment → target app, route in app.main
   - call the node via smartasync
   - build_wsx_response(id, status, headers, data) → ws.send_text()

NATS (roadmap — not implemented)

A NATS transport would follow the same request → handler → response shape, using the native reply-to subject for routing and the id field for application tracing. No NATS code exists in the package today.

Connection Authentication

When a WSX WebSocket connects, WsxHandler._authenticate(scope) authenticates the handshake (using the connection’s headers, parsed into scope["_headers"]).

It does not depend on a hard reference to the auth middleware: it walks the server middleware chain starting at server.dispatcher and following each layer’s .app attribute, looking for an object that exposes both _authenticate and _auth_config (i.e. AuthMiddleware). If found, its _authenticate(scope) result is used; otherwise authentication returns None.

The result (when authenticated) is a dict shaped like {"tags": [...], "identity": "...", "backend": "bearer|basic|jwt:..."}. The handler then:

stores it as scope["auth"],
exposes scope["_filters"]["auth_tags"] (the tags list, or []),
passes it to WsxRegistry.register so the connection is queryable by identity.

Each routed message reuses these filters: MsgRequest reads scope["_filters"]["auth_tags"] and the router enforces them via auth_tags= when resolving the node.

Inline Control Channel (`_wsx/`)

Messages whose path starts with _wsx/ are control messages: they are handled inline by WsxHandler._handle_control and are never routed to an app.

Currently implemented:

Control path	Behaviour
`_wsx/ping`	replies with `build_wsx_response(id=..., status=200, data="pong")`
any other `_wsx/...`	replies `status=404`, `data={"error": "Unknown control path: ..."}`

Example exchange:

client → WSX://{"id":"k1","method":"GET","path":"_wsx/ping"}
server → WSX://{"id":"k1","status":200,"data":"pong"}

Module Structure

The wsx/ package contains protocol, handler, and connection-registry code:

src/genro_asgi/wsx/
├── __init__.py     # public exports
├── protocol.py     # WSX_PREFIX, is_wsx_message, parse_wsx_message,
│                   #   build_wsx_message, build_wsx_response (stdlib json only)
├── handler.py      # WsxHandler — accept, authenticate, receive loop, dispatch
└── registry.py     # WsxRegistry, WsxConnectionInfo — track/lookup/broadcast

Request classes are in request.py (MsgRequest), not in wsx/. TYTX hydration also lives in request.py, not in wsx/protocol.py.

Public exports (wsx/__init__.py __all__): WSX_PREFIX, is_wsx_message, parse_wsx_message, build_wsx_message, build_wsx_response, WsxHandler, WsxRegistry, WsxConnectionInfo.

Protocol API (`wsx/protocol.py`)

The protocol module is intentionally minimal and uses stdlib json only.

WSX_PREFIX = "WSX://"


def is_wsx_message(data: str | bytes) -> bool:
    """True if data starts with the WSX:// prefix (str or bytes)."""
    if isinstance(data, bytes):
        return data.startswith(b"WSX://")
    return data.startswith(WSX_PREFIX)


def parse_wsx_message(data: str | bytes) -> dict[str, Any]:
    """Parse a WSX message into a dict using stdlib json.

    Strips the WSX:// prefix if present. Does NOT perform TYTX hydration —
    TYTX decoding lives in MsgRequest._parse_wsx_message (request.py).
    """
    if isinstance(data, bytes):
        data = data.decode("utf-8")
    if data.startswith(WSX_PREFIX):
        data = data[len(WSX_PREFIX):]
    return dict(json.loads(data))


def build_wsx_message(
    *,
    id: str,
    method: str,
    path: str = "/",
    headers: dict[str, str] | None = None,
    cookies: dict[str, str] | None = None,
    query: dict[str, Any] | None = None,
    data: Any = None,
    tytx: bool = False,
) -> str:
    """Build a WSX:// request string (stdlib json.dumps).

    Optional fields are included only when truthy/non-None. When tytx=True a
    "tytx": true marker is added so the receiver hydrates types.
    """
    ...


def build_wsx_response(
    *,
    id: str,
    status: int = 200,
    headers: dict[str, str] | None = None,
    cookies: dict[str, Any] | None = None,
    data: Any = None,
) -> str:
    """Build a WSX:// response string (stdlib json.dumps).

    Always emits id and status; adds headers/cookies/data only when present.
    """
    ...

Note: there is no serialize_wsx_response function and no ::JS / to_tytx handling in protocol.py. The response serializer is build_wsx_response, and TYTX is applied (on input) only by request.py.

Error Handling

WsxHandler._dispatch_message turns exceptions into a WSX response with the matching status, keeping the request’s id:

HTTPException → build_wsx_response(id, status=exc.status_code, data={"error": exc.detail})
any other Exception → status=500, data={"error": str(exc)} (and a logged traceback)

WSX://{
    "id": "uuid-123",
    "status": 404,
    "data": {"error": "User not found"}
}

The router error names (not_found, not_authorized, not_authenticated, not_available, validation_error) are all mapped to HTTPException, the same mapping used by the HTTP Dispatcher.

Broadcast and Directed Sends

WsxRegistry (in wsx/registry.py) tracks every live connection as a WsxConnectionInfo (connection_id, websocket, scope, auth; with identity and client properties). It supports:

register(connection_id, websocket, scope, auth) / unregister(connection_id)
get(connection_id) — lookup by id
find_by_identity(identity) — all connections for an authenticated identity
broadcast(data, *, exclude=None) — send a WSX response (id="_broadcast", status=200) to every connection except exclude; returns the count sent
send_to(identity, data) — send a WSX response (id="_directed", status=200) to all connections of an identity; returns the count sent
len(registry), iter(registry), connection_id in registry

Streaming (roadmap — not implemented)

A streaming convention (multiple responses sharing one id, with a stream boolean marking continuation/end) is envisaged but not implemented: build_wsx_response has no stream parameter and the handler sends exactly one response per request.

Correlation ID

WebSocket: required in the WSX message id field; the response reuses it. The client id becomes MsgRequest.external_id, while the server assigns its own internal id (a fresh uuid4).
HTTP: generated by the server (or taken from an x-request-id header).
NATS (roadmap): native reply-to for routing, id for application tracing.

Transport-Agnostic Handler

Handlers receive a BaseRequest and work the same regardless of transport. For WSX, request.transport returns "websocket".

async def get_user(request: BaseRequest) -> dict:
    """Works for HTTP and WSX/WebSocket alike."""
    user_id = request.path.split("/")[-1]
    auth = request.headers.get("authorization")
    session = request.cookies.get("session_id")

    user = await db.get_user(user_id)
    return {
        "id": user.id,
        "name": user.name,
        "transport": request.transport,  # "http" or "websocket"
    }