Transports
The muxws peer does not open sockets. It uses a SocketAdapter with four methods (send_text, send_bytes, receive, close), and the adapters are the only transport-specific code in the library. The rest of the library, in both languages, uses only those four methods and has no WebSocket-specific code.
Shipped adapters: Starlette/FastAPI and the websockets library in Python, the browser's global WebSocket and the ws package in TypeScript, and an in-memory pair in each language for tests.
Every snippet below is a complete file and runs as shown, unless it is marked # fragment / // fragment.
FastAPI / Starlette acceptor
import uvicorn
from fastapi import FastAPI, WebSocket
from muxws import Stream, serve
app = FastAPI()
async def on_stream(payload, stream: Stream) -> None:
"""One handler for every inbound stream on this connection."""
if payload == {"op": "count"}:
for n in range(3):
await stream.send({"n": n})
await stream.end()
return
await stream.reply({"echo": payload})
@app.websocket("/ws")
async def muxws_endpoint(websocket: WebSocket) -> None:
# Do NOT call websocket.accept() first: muxws performs the upgrade, because it selects the
# subprotocol.
await serve(websocket, handler=on_stream)
if __name__ == "__main__":
uvicorn.run(app, host="127.0.0.1", port=8000)Needs pip install "muxws[starlette]" fastapi uvicorn.
Do not accept the WebSocket in the application. accept(), and serve(), which calls it, performs the upgrade, because selecting the muxws.v1.<codec> subprotocol is part of the upgrade. If the application has already accepted the WebSocket, accept() raises ProtocolError.
If the offered codec does not match the acceptor's, the upgrade is denied with HTTP 400 and no accept takes place. accept() then raises CodecMismatch; serve() catches it and returns without an error, because the 400 response has already been sent.
Use accept() instead of serve() when the application needs the peer object, for example to tag it, register it or push to it:
# fragment
from muxws import PeerRegistry, accept
registry = PeerRegistry()
@app.websocket("/ws")
async def muxws_endpoint(websocket: WebSocket) -> None:
peer = await accept(websocket)
peer.on_stream(on_stream)
peer.tags["tenant"] = websocket.query_params.get("tenant", "")
with registry.registered(peer):
await peer.serve()websockets acceptor, with select_subprotocol
import asyncio
import websockets
from muxws import Stream, accept, select_subprotocol
async def on_stream(payload, stream: Stream) -> None:
await stream.reply({"echo": payload})
async def handle(connection) -> None:
peer = await accept(connection)
peer.on_stream(on_stream)
await peer.serve()
async def main() -> None:
async with websockets.serve(
handle,
"127.0.0.1",
8000,
select_subprotocol=select_subprotocol,
) as service:
print(f"listening on {service.sockets[0].getsockname()}")
await asyncio.Future()
asyncio.run(main())Needs pip install "muxws[websockets]".
select_subprotocol is installed as the websockets handshake hook and is called with (connection, subprotocols). Pass the function itself; do not call it, and do not wrap it.
It raises rather than returning None
muxws.select_subprotocol raises NegotiationError on a mismatch. A hook that returns None causes websockets to answer 101 without a Sec-WebSocket-Protocol header: a successful handshake with no subprotocol selected, and the mismatch is then detected only on the open socket. websockets converts an InvalidHandshake raised by this hook into HTTP 400 and any other exception into 500, so NegotiationError produces the 400 the protocol requires.
To install your own hook, call muxws's from it instead of reimplementing it:
def select(connection, subprotocols): # fragment
# application checks first; then delegate, and let it raise
return select_subprotocol(connection, subprotocols)accept() takes the websockets connection directly: it recognises the type, checks that the negotiated subprotocol matches this peer's codec, and wraps it. The handshake is already complete when the handler is called, so this check cannot produce a 400; only the hook above can. On a mismatch it logs one error line, closes the socket with 1008 and raises CodecMismatch.
websockets dialer
import asyncio
from muxws import Reconnect, connect
async def main() -> None:
peer = await connect(
"ws://127.0.0.1:8000/ws",
hello={"client": "reporting"},
reconnect=Reconnect(),
ping_interval=20.0, # seconds
ping_timeout=10.0, # seconds
)
print(await peer.request({"hello": "world"}, timeout=5.0)) # seconds
await peer.close()
asyncio.run(main())Needs pip install "muxws[websockets]". It can dial any of the acceptors on this page. Without the extra, the first connect() raises muxws.transports.websockets_.WebsocketsNotInstalledError, whose message contains that install command.
connect() uses websockets; Python has no other dialer and no adapter option. It offers muxws.v1.<codec> first in its subprotocol list, verifies the selected subprotocol, and passes the connection to a Peer that is already serving. Application subprotocol entries go in subprotocols=[...], after the muxws entry; the acceptor ignores them. The URL alone selects the underlying socket: a ws+unix:// URL dials a socket file through the same function, with the same handshake and the same options; see Unix domain sockets.
connect() raises if the first attempt fails, regardless of reconnect=. See Reconnect.
Browser dialer
import { connect, type Stream } from 'muxws';
const peer = await connect('wss://example.test/ws', {
hello: { tab: sessionStorage.getItem('tab') },
pingIntervalMs: 20_000, // milliseconds
pingTimeoutMs: 10_000, // milliseconds
onStream: async (payload: unknown, stream: Stream) => {
// Streams the server opens (server push) arrive here.
console.log('pushed', payload);
await stream.end();
},
onReconnect: (attempt: number) => {
console.log(`reconnected (${attempt}); resubscribing`);
},
});
console.log(await peer.request({ hello: 'world' }, { timeoutMs: 5000 })); // millisecondsNeeds npm install muxws only: the browser entry point has no runtime dependencies, and BrowserSocket, the adapter over the platform's global WebSocket, is the only WebSocket-specific file in it.
onStream is passed to connect() rather than registered after it returns. The acceptor may open a stream as soon as it receives the hello, and a peer whose handler is registered one await later answers that stream with reset(REFUSED, "no on_stream handler"). The same applies to onClose and onReconnect.
A browser cannot set request headers on a WebSocket handshake, so this connect() has no headers option. Credentials are sent as a cookie, in the URL, or as an extra subprotocol entry; see below.
A browser cannot open a filesystem socket, so this connect() rejects a ws+unix: URL immediately with a message that names muxws/node. See Unix domain sockets.
Node ws acceptor, via muxws/node
import { WebSocketServer } from 'ws';
// `muxws/node` does not register the JSON codec; the main entry point does, on import.
import 'muxws';
import { handleProtocols, refuseMismatchedUpgrade, serve } from 'muxws/node';
import type { Stream } from 'muxws';
async function onStream(payload: unknown, stream: Stream): Promise<void> {
await stream.reply({ echo: payload });
}
// Both hooks. `handleProtocols` selects; `refuseMismatchedUpgrade` refuses.
const server = refuseMismatchedUpgrade(new WebSocketServer({ port: 8000, handleProtocols }));
server.on('connection', (socket) => {
void serve(socket, { handler: onStream }).catch((error: unknown) => {
console.error('connection ended', error);
});
});Needs npm install muxws ws. ws is an optional peer dependency, imported only from this subpath, so a browser bundle does not contain it. It is imported dynamically and only on the dial path, so a process that only accepts connections runs without it; a process that dials without it gets WsNotInstalledError from muxws/node, a TransportUnsupportedError whose message is npm install ws.
The snippet has three requirements:
import 'muxws'; is required. The main entry point registers the JSON codec, and muxws/node does not, because a module with import-time side effects cannot be removed by tree-shaking; the registration is in the entry point a browser loads. A Node acceptor that imports only from muxws/node fails on its first connection with CodecNotRegistered: codec 'json' is not registered … registered codecs are []. Alternatively, call registerCodec('json', new JsonCodec()) from muxws during bootstrap.
Both hooks are required. handleProtocols only selects a value: when it returns false, ws still answers 101, without a Sec-WebSocket-Protocol header. refuseMismatchedUpgrade wraps the server's shouldHandle, the hook that can reject an upgrade with a status, so a mismatched codec receives HTTP 400. A server that installs only handleProtocols completes handshakes that should be refused. The wrapper calls the inherited shouldHandle first, so a server constructed with path keeps that check.
Without refuseMismatchedUpgrade, the upgrade is answered with 101 instead of 400 and the connection completes. accept() then finds no matching subprotocol on the open socket, logs one error line, closes the socket with 1008 and throws CodecMismatch; serve() resolves. The dialer's error depends on its language: a Python dialer detects the mismatch on the open socket and raises CodecMismatch, while a browser or Node dialer is rejected by its own WebSocket client before muxws runs, with a message such as Server sent no subprotocol that names neither codec. If a cross-language dial fails this way, check this hook first.
Use accept() to get the peer. serve is accept plus a handler plus the read loop; accept(socket, options) returns the Peer for tagging and registration.
Unix domain sockets
A muxws connection can be dialled over a socket file instead of a TCP port. The protocol is the same WebSocket: the same HTTP GET + Upgrade, the same muxws.v1.<codec> subprotocol answered 101 or 400, carried by the same WebsocketsSocket and WsSocket adapters. Frames, close codes and connect() arguments are unchanged. Only the URL differs:
ws+unix:///run/muxws/api.sock:/ws
└────────┬────────┘ └┬┘
socket file request targetThe URL's path, including the query, is split on the first colon. The part before it is the filesystem path to open; the part after it is the HTTP request target for the GET line of the upgrade, and must begin with /. Both ports reject a target that does not, before opening a socket. Without that check, an unrooted target would be merged into the authority: …api.sock:ws would become ws://localhostws, and the dial would open the correct file and request / with an unintended Host header. An acceptor that does not route on the request target (unix_serve, or a WebSocketServer without path) would accept that handshake.
With no colon, or with a trailing colon and nothing after it, the request target is /. The query is part of the split string, so it belongs to the target when a colon precedes it (ws+unix:///run/muxws/api.sock:/ws?tenant=42 opens that file and requests /ws?tenant=42) and to the filename when no colon precedes it. An authority between the // and the path is used only as the Host header; without an authority, the header is localhost. The scheme is case-insensitive, and a single slash (ws+unix:/run/muxws/api.sock:/ws) parses the same way as two.
The socket path is taken from the URL as written: it is neither percent-encoded nor percent-decoded. ws+unix:///tmp/my dir/api.sock opens /tmp/my dir/api.sock, and ws+unix:///tmp/a%20b.sock opens a file whose name contains the three characters %20. Both ports apply this rule.
There is no wss+unix://. A socket file cannot be reached from another machine and has no host name for certificate verification, so it is protected by its directory's permissions, not by TLS. Dialling wss+unix: raises UnixUrlError; it is not downgraded to plaintext.
The dialer is the public connect(), unchanged:
"""Unix-domain-socket dialer: the ordinary public `connect()`, pointed at a socket file.
Start `uds_server.py` first, then run `python uds_client.py`. The URL is
`ws+unix://<tempdir>/muxws.sock:/ws` unless the `MUXWS_URL` environment variable says otherwise.
`connect()` selects the transport from the URL scheme; no extra argument or adapter is needed. The
codec offer at the handshake, the `CodecMismatch` raised for HTTP 400, the heartbeat and the reconnect
loop use the same code as over TCP. The path is split at the first colon: `/tmp/muxws.sock` is the
file to open and `/ws` is the HTTP request target of the upgrade. There is no `wss+unix://`: a
filesystem socket is not reachable from another machine, and the socket file's permissions control
access.
"""
import asyncio
import os
import tempfile
from muxws import connect
DEFAULT_SOCKET = os.path.join(tempfile.gettempdir(), "muxws.sock")
URL = os.environ.get("MUXWS_URL", f"ws+unix://{DEFAULT_SOCKET}:/ws")
async def main() -> None:
peer = await connect(URL)
try:
# The quick start's two call shapes, unary and streaming, over a Unix socket.
answer = await peer.request({"say": "hello"})
print(f"greeting: {answer['greeting']}")
stream = peer.open({"count": 3}, end=True)
async for chunk in stream:
print(f"chunk {chunk['chunk']} of {chunk['of']}")
# The heartbeat is a muxws `ping` frame, sent over the same socket. Only the arrival of the pong
# is printed: the round-trip time varies between runs, and the guide compares this output
# exactly.
await peer.ping()
print("ping: answered")
finally:
await peer.close()
if __name__ == "__main__":
asyncio.run(main())The acceptor is websockets.asyncio.server.unix_serve with the same select_subprotocol hook the TCP acceptor installs, and the same accept() afterwards:
"""Unix-domain-socket acceptor: `unix_serve`, one muxws peer, and the caller named by the kernel.
Run it with `python uds_server.py`. It binds `<tempdir>/muxws.sock` unless the `MUXWS_SOCKET`
environment variable says otherwise. The documentation test sets it to a path in a short temporary
directory: a Unix socket path is limited to about 108 bytes, and `bind()` raises
`OSError: AF_UNIX path too long` for a longer one.
The muxws code is the same as over TCP. `unix_serve` performs the same HTTP GET + Upgrade as
`websockets.serve`, `select_subprotocol` selects the same `muxws.v1.<codec>` value, `accept()` wraps
the connection in the same `WebsocketsSocket`, and a dialer offering an unsupported codec is refused
with HTTP 400 before anything is accepted. Only the underlying socket differs.
The Unix socket adds the `accepted a dialer:` line. The socket file's permissions decide who may
connect, and `SO_PEERCRED` gives the acceptor the caller's pid, uid and gid from the kernel. The
connection is authenticated at the upgrade, before `accept()`, without a credential on the wire.
"""
import asyncio
import os
import socket
import stat
import struct
import tempfile
from contextlib import closing
from websockets.asyncio.server import unix_serve
from muxws import accept, select_subprotocol, Stream
#: The socket file this acceptor binds. Keep it short; see the 108-byte limit above.
SOCKET_PATH = os.environ.get("MUXWS_SOCKET", os.path.join(tempfile.gettempdir(), "muxws.sock"))
async def on_stream(payload: object, stream: Stream) -> None:
"""The incoming-stream handler; the same as the TCP quick start's, apart from the greeting."""
if isinstance(payload, dict) and payload.get("say") == "hello":
# Unary: one payload and the end of the stream in a single call.
await stream.reply({"greeting": "hello over a unix socket"})
return
if isinstance(payload, dict) and isinstance(payload.get("count"), int):
# Streaming response: any number of payloads, then an end.
total = payload["count"]
for n in range(1, total + 1):
await stream.send({"chunk": n, "of": total})
await stream.end()
return
# Any other stream is acknowledged by returning: muxws ends a stream its handler left open.
def peer_credentials(connection: object) -> str:
"""The dialer's pid, uid and gid as the kernel reports them, or why they are unavailable.
`SO_PEERCRED` is a Linux socket option. macOS and the BSDs provide the uid and gid, but not the
pid, through `getpeereid()`, which Python does not expose; no other muxws transport provides peer
credentials. Where they are unavailable this returns a message instead of raising, so the example
runs on those platforms.
"""
transport = getattr(connection, "transport", None)
raw = transport.get_extra_info("socket") if transport is not None else None
if raw is None or not hasattr(socket, "SO_PEERCRED"):
return "unavailable on this platform"
layout = "3i" # struct ucred: pid_t, uid_t, gid_t, three 32-bit integers.
credentials = raw.getsockopt(socket.SOL_SOCKET, socket.SO_PEERCRED, struct.calcsize(layout))
pid, uid, gid = struct.unpack(layout, credentials)
return f"pid={pid} uid={uid} gid={gid}"
async def handle(connection: object) -> None:
"""One connection, one peer. `unix_serve` calls this once per dialer, as `serve` does."""
print(f"accepted a dialer: {peer_credentials(connection)}", flush=True)
peer = await accept(connection)
peer.on_stream(on_stream)
await peer.serve()
def is_stale(path: str) -> bool:
"""True when `path` is a socket file with no listener, as left behind by a killed acceptor.
A socket file remains after the process that bound it exits, and the next `bind()` fails with
`EADDRINUSE` until it is removed. Removing it unconditionally would detach a running acceptor
from its path. A connection attempt distinguishes the cases: it is refused when no process
listens, and accepted by a running acceptor.
The file type is checked before the connection attempt, because `connect()` on a regular file or
a directory also fails with `ECONNREFUSED`. A path that is not a socket exits with an error and is
never removed.
"""
if not os.path.exists(path):
return False
if not stat.S_ISSOCK(os.lstat(path).st_mode):
raise SystemExit(f"{path} exists and is not a socket file; refusing to remove it")
with closing(socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)) as probe:
return probe.connect_ex(path) != 0
async def main() -> None:
if not hasattr(socket, "AF_UNIX"):
raise SystemExit("this example needs AF_UNIX, which Windows does not have; use the TCP quick start")
if is_stale(SOCKET_PATH):
os.unlink(SOCKET_PATH)
async with unix_serve(handle, SOCKET_PATH, select_subprotocol=select_subprotocol):
print(f"listening on {SOCKET_PATH}", flush=True)
await asyncio.Future()
if __name__ == "__main__":
asyncio.run(main())Both need pip install "muxws[websockets]". Start the acceptor, then run the dialer; the dialer prints:
Running the pair without two terminals
python demo.py --uds from the repository root starts this acceptor on a socket file in a temporary directory, runs this dialer against it, and removes both afterwards. It runs these two files, not copies of them.
greeting: hello over a unix socket
chunk 1 of 3
chunk 2 of 3
chunk 3 of 3
ping: answeredThese are the quick start's two call shapes, one unary request and one streamed response, over a socket file instead of a port, followed by a ping round trip.
The Node acceptor uses { server } rather than { port }. ws cannot bind a socket file, so an http.Server binds it and the WebSocketServer attaches to that server's upgrade event. refuseMismatchedUpgrade wraps shouldHandle in both configurations, so a codec mismatch still receives HTTP 400:
import { statSync, unlinkSync } from 'node:fs';
import { createServer } from 'node:http';
import { connect as netConnect } from 'node:net';
import { WebSocketServer } from 'ws';
// Required: the main entry point registers the JSON codec, and `muxws/node` does not.
import 'muxws';
import { handleProtocols, refuseMismatchedUpgrade, serve } from 'muxws/node';
import type { Stream } from 'muxws';
const SOCKET = process.env.MUXWS_SOCKET ?? '/tmp/muxws.sock';
async function onStream(payload: unknown, stream: Stream): Promise<void> {
await stream.reply({ echo: payload });
}
/**
* Remove the socket file only if it is stale, left by a process that exited without closing it.
*
* A socket file remains after the process that bound it exits, and `listen()` on the path fails with
* EADDRINUSE until it is removed. Unlinking unconditionally would remove the address of a running
* acceptor, and a path that is not a socket must not be removed: reject a non-socket path, probe a
* socket path, and unlink it only if the probe connection is refused.
*/
async function clearStaleSocket(path: string): Promise<void> {
let info;
try {
info = statSync(path);
} catch {
return; // No file at the path.
}
if (!info.isSocket()) throw new Error(`${path} exists and is not a socket file; refusing to remove it`);
const alive = await new Promise<boolean>((resolve) => {
const probe = netConnect(path);
probe.on('connect', () => {
probe.destroy();
resolve(true);
});
probe.on('error', () => resolve(false));
});
if (alive) throw new Error(`${path} already has an acceptor listening on it`);
unlinkSync(path);
}
await clearStaleSocket(SOCKET);
const httpServer = createServer();
const server = refuseMismatchedUpgrade(new WebSocketServer({ server: httpServer, handleProtocols }));
server.on('connection', (socket) => {
void serve(socket, { handler: onStream }).catch((error: unknown) => {
console.error('connection ended', error);
});
});
// SIGINT's default action ends the process without running `close()` and leaves the socket file,
// which `clearStaleSocket` removes on the next start.
process.on('SIGINT', () => {
httpServer.close(() => process.exit(0));
});
// Listen on the HTTP server, not on the WebSocketServer: only `http.Server` has a `listen()` that
// binds a path.
httpServer.listen(SOCKET, () => {
console.log(`listening on ${SOCKET}`);
});The Node dialer is also the same connect(), with the URL passed unchanged:
import 'muxws';
import { connect } from 'muxws/node';
const peer = await connect('ws+unix:///tmp/muxws.sock:/ws');
console.log(await peer.request({ say: 'hello' }, { timeoutMs: 5000 })); // milliseconds
await peer.close();muxws/node splits on the first colon itself. ws supports ws+unix:, but it splits the path on every colon and keeps the second field: given the whole URL, it would dial ws+unix:///run/api.sock:/ws:v2 with the target /ws and ws+unix:///run/api.sock:/ws?since=2026-08-14T10:00:00Z with /ws?since=2026-08-14T10, while Python requests the whole target for both URLs. Neither end would detect the truncation: the acceptor routes on the target it receives, and the dialer does not report it. muxws/node therefore performs the split and passes ws a plain ws: URL plus a connector that opens the file. ws still performs the dial, and everything after the split is handled by ws: the subprotocol offer, the events, the HTTP 400.
The browser entry point rejects this scheme: neither a browser nor Node's global WebSocket can open a filesystem socket, so connect() from muxws rejects a ws+unix: URL immediately with a UnixSocketsUnsupportedError naming muxws/node. It is a TransportUnsupportedError: the URL is valid, and the application must dial through a different entry point.
On POSIX, both ports raise the same class for the same three invalid URLs. A ws+unix: URL with a request target that does not begin with /, a ws+unix: URL naming no socket file, and a wss+unix: URL raise UnixUrlError: from muxws.transports.unix import UnixUrlError in Python, import { UnixUrlError } from 'muxws/node' in TypeScript. Both are TransportUrlErrors, so except TransportUrlError / instanceof TransportUrlError catches an invalid address for any transport, ws:// included, without importing a transport module. See Errors.
A Unix domain socket identifies the caller. The socket file has an owner, a group and a mode, so the kernel checks who may open the connection before any HTTP is sent; placing the socket in a directory that only the intended callers can traverse restricts access with chmod. Once a connection is open, SO_PEERCRED on Linux gives the acceptor the caller's pid, uid and gid from the kernel. The remote process cannot forge these values, and they are not sent over the connection. This is authentication at the upgrade, before accept(), as described in Where authentication belongs, with no credential on the wire and no cookie, header or token to rotate. uds_server.py above reads it on Linux and prints unavailable on this platform elsewhere: macOS and the BSDs have getpeereid(), which returns the uid and the gid but not the pid and which CPython does not expose, and Node's standard library exposes neither call. On those platforms, the file permissions are the only access control.
Path length is limited. A socket path is copied into sockaddr_un.sun_path, which is about 108 bytes on Linux and smaller on some other systems, so keep both the directory and the basename short. An overlong path raises an error (CPython raises OSError: AF_UNIX path too long, and libuv returns EINVAL with the whole path), but neither message names the component to shorten. A path built from a long TMPDIR commonly exceeds the limit.
AF_UNIX does not exist on Windows, and the two ports behave differently there. In Python, connect() raises muxws.transports.unix.UnixSocketsUnsupportedError (a TransportUnsupportedError, so both a MuxwsError and a RuntimeError) from the URL parse, before opening a socket and before the first attempt, so a background reconnection never raises it. This check runs before the two URL shape checks, so on Windows an invalid request target and a URL naming no socket file also raise UnixSocketsUnsupportedError; only wss+unix: raises UnixUrlError on every platform. muxws/node raises no equivalent error: on Windows, net.connect({ path }) opens a named pipe, but no ws+unix: URL can address one, because new URL() rejects the backslashes in \\.\pipe\name on every platform. A Windows dial of a POSIX-style path therefore fails with a connect error naming the path.
In-memory, for tests
Both languages ship a pair of adapters connected to each other without a socket. The library's tests and the conformance runner use them to run two peers.
from muxws import Peer, get_codec
from muxws.transports.memory import memory_pair
left_socket, right_socket = memory_pair()
dialer = Peer(left_socket, codec=get_codec("json"), is_dialer=True)
acceptor = Peer(right_socket, codec=get_codec("json"), is_dialer=False)import { Peer, getCodec, memoryPair } from 'muxws';
const [leftSocket, rightSocket] = memoryPair();
const dialer = new Peer(leftSocket, { codec: getCodec('json'), isDialer: true });
const acceptor = new Peer(rightSocket, { codec: getCodec('json'), isDialer: false });The memory pair needs no handshake: a muxws connection is established when the socket is open with the subprotocol accepted, and a memory pair is created in that state.
Where authentication belongs
At the upgrade, before accept(). muxws does not interpret credentials anywhere.
The upgrade is an ordinary HTTP request, so use HTTP mechanisms: a cookie, an Authorization header, a signed query parameter, a mutual-TLS client certificate, or, for browsers, an extra subprotocol entry, because a browser can set subprotocols but not headers.
# fragment
@app.websocket("/ws")
async def muxws_endpoint(websocket: WebSocket) -> None:
identity = authenticate(websocket.headers, websocket.cookies)
if identity is None:
# Closed before accept: the upgrade is denied and no muxws frame is sent.
await websocket.close(code=1008)
return
peer = await accept(websocket)
peer.tags["user"] = identity.user_id
peer.on_stream(on_stream)
await peer.serve()Not in the hello. The hello is an ordinary stream: it arrives after the socket is open and the acceptor has allocated resources for it, it is sent unchanged on every reconnection for the life of the process (so the credential cannot be rotated), and it is passed as an application payload to every frame observer on the connection. The hello identifies a client; it does not authorize the connection. See Reconnect.
Not in per-stream headers. muxws does not interpret them. They carry application metadata, such as a trace id, a content type or an idempotency key. Authenticating per stream would re-check a credential on every stream of a socket that was authenticated at the upgrade, and muxws provides no support for it. muxws also does not route on the opening payload: it has no path matching, no method dispatch and no handler table. Each peer has one on_stream handler, and the application decides how to handle each payload.
When a credential expires during a connection, close the connection with goaway. Streams at or below the cut-off finish, no new streams are accepted, and a dialer with a reconnect helper dials again and authenticates at the next upgrade.
# fragment
await peer.close(reason="credential expired")Writing your own adapter
An adapter implements four methods. No other part of the library changes.
class MySocket: # fragment
async def send_text(self, text: str) -> None: ...
async def send_bytes(self, data: bytes) -> None: ...
async def receive(self) -> str | bytes: ...
async def close(self, code: int = 1000, reason: str = "") -> None: ...interface SocketAdapter { // fragment
sendText(text: string): Promise<void> | void;
sendBytes(bytes: ArrayBuffer): Promise<void> | void;
receive(): Promise<string | ArrayBuffer>;
close(code?: number, reason?: string): Promise<void> | void;
}Two requirements are not visible in the signatures:
receive()must raiseConnectionClosedwhen the socket dies; it must not return or block. The exception tells the peer that the connection ended and starts socket-death handling: every live stream is failed,on_closefires, and the reconnect helper is notified.- Text and binary are separate methods. The peer chooses between them from the codec's declared
binaryflag and does not inspect the data it sends. A singlesendmethod that accepted both would hide a mismatch between a codec's declaration and its output.
In Python, pass an adapter instance to accept(), which accepts any object implementing the interface. In TypeScript, accept() takes a ws connection and wraps it in WsSocket, so a custom adapter is passed directly to the peer: new Peer(adapter, { codec, isDialer: false }).
See also
api/transports—SocketAdapter, the Starlette adapter, thewebsocketsadapter, the in-memory adapterapi/accept—accept,serve,select_subprotocolapi/connect—connectandConnectOptionsapi/peer—on_stream,serve,tags,closeapi/codec—registerCodecand the codec the subprotocol namesapi/errors—CodecMismatch,CodecNotRegistered,ConnectionClosed