Getting Started
Installation
muxws has no required runtime dependencies in either language. A transport, a codec or a web framework is something you already have, or something you ask for by name.
# Python. The extras are transports and codecs, and each one is optional:
# muxws[starlette] - the FastAPI / Starlette acceptor
# muxws[websockets] - the `websockets` dialer and acceptor
# muxws[msgpack] - the msgpack codec
# The quick start below runs a FastAPI server under uvicorn, so it asks for those two as well.
pip install "muxws[starlette,websockets]" fastapi uvicorn# TypeScript. `ws` and `@msgpack/msgpack` are optional peer dependencies: `ws` is needed only for a
# Node dialer or acceptor (a browser has WebSocket already), `@msgpack/msgpack` only for the msgpack
# codec. `tsx` is here to run the .ts file below without a build step.
npm install muxws ws
npm install --save-dev tsx
# The client below is an ES module - it uses `await` at the top level - and the package.json npm
# just wrote for you does not say so. Without this line `npx tsx` refuses the file with
# "Top-level await is currently not supported with the cjs output format".
npm pkg set type=moduleQuick Start
Three files, one socket, two call shapes: a unary request that gets exactly one answer, and a streaming response that arrives in pieces. The server is Python; the client is Python or TypeScript, and both print the same thing, because both speak the same wire.
1. The server
The route hands the socket to accept() and then does nothing transport-specific ever again. accept() performs the WebSocket upgrade itself, because it is the only party that knows which muxws.v1.<codec> subprotocol to select.
There is one on_stream handler per peer, and every stream the other end opens arrives at it — the first one, and every one after that. stream.reply() is one payload plus the end of the stream; stream.send() as many times as you like, then stream.end(), is the streaming shape.
"""Quick-start acceptor: FastAPI, one WebSocket route, one muxws peer.
Run it with `python quickstart_server.py`. It listens on 127.0.0.1:8000 unless the `MUXWS_PORT`
environment variable says otherwise - the documentation test starts it on a free port that way.
The route does three things and nothing else: hand the socket to `accept()`, register the one
`on_stream` handler this peer has, and run the read loop. Every stream the dialer opens - now or an
hour from now - arrives at that handler.
"""
import os
from fastapi import FastAPI, WebSocket
from muxws import accept, Stream
#: 127.0.0.1 on purpose: the quick start is a local demonstration, not a deployment.
HOST = "127.0.0.1"
PORT = int(os.environ.get("MUXWS_PORT", "8000"))
app = FastAPI()
async def on_stream(payload: object, stream: Stream) -> None:
"""The one incoming-stream handler. `payload` is the value the dialer opened the stream with."""
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, muxws"})
return
if isinstance(payload, dict) and isinstance(payload.get("count"), int):
# Streaming response: as many payloads as we like, then an end. `end` is a flag on the last
# frame, not a frame of its own, so this costs three frames rather than four.
total = payload["count"]
for n in range(1, total + 1):
await stream.send({"chunk": n, "of": total})
await stream.end()
return
# Anything else - the reconnect hello of the last section, for one - is acknowledged simply by
# returning: muxws ends a stream its handler left open.
@app.websocket("/ws")
async def muxws_endpoint(websocket: WebSocket) -> None:
"""`accept()` performs the WebSocket upgrade itself, because it is what selects the subprotocol."""
peer = await accept(websocket)
peer.on_stream(on_stream)
await peer.serve()
if __name__ == "__main__":
import uvicorn
uvicorn.run(app, host=HOST, port=PORT, log_level="warning")Run it:
python quickstart_server.pyIt listens on 127.0.0.1:8000. Set MUXWS_PORT to move it.
2. The Python client
connect() returns a peer that is already serving. request() is the unary shape: one payload out, exactly one payload back. open() is the general one — it is synchronous, it hands back the Stream in the same turn it allocated the id, and end=True says this side has nothing more to send.
"""Quick-start dialer: the `websockets` transport, two call shapes on one socket.
Start `quickstart_server.py` first, then run `python quickstart_client.py`. The URL is
`ws://127.0.0.1:8000/ws` unless the `MUXWS_URL` environment variable says otherwise.
"""
import asyncio
import os
from muxws import connect
URL = os.environ.get("MUXWS_URL", "ws://127.0.0.1:8000/ws")
async def main() -> None:
peer = await connect(URL)
try:
# Unary. `request()` sends one payload, waits for exactly one back, and raises if the remote
# sends a second - which `await stream` deliberately does not.
answer = await peer.request({"say": "hello"})
print(f"greeting: {answer['greeting']}")
# Streaming response. `open()` is synchronous: it hands back the Stream in the same turn it
# allocated the id. `end=True` says this side has nothing more to send, so the acceptor is
# free to stream back immediately.
stream = peer.open({"count": 3}, end=True)
async for chunk in stream:
print(f"chunk {chunk['chunk']} of {chunk['of']}")
finally:
await peer.close()
if __name__ == "__main__":
asyncio.run(main())Run it in a second terminal:
python quickstart_client.py3. The TypeScript client
The same two calls, against the same server. Two imports: muxws is the package, and importing it is what registers the built-in json codec; muxws/node is the subpath that owns the ws socket, and it is the only part of the package that touches ws. A browser dialer imports connect from muxws instead and changes nothing else.
Note the units: every duration in the TypeScript port is milliseconds as an integer, where its Python twin is seconds as a float. Neither call below takes one, but that is the rule when you reach for request(..., { timeoutMs }) or result({ timeoutMs }).
/**
* Quick-start dialer in TypeScript, on Node, against the same `quickstart_server.py`.
*
* Run it with `npx tsx quickstart-client.ts`. The URL is `ws://127.0.0.1:8000/ws` unless the
* `MUXWS_URL` environment variable says otherwise.
*
* Two imports, and both are load-bearing. `muxws` is the package itself, and importing it is what
* registers the built-in `json` codec; `muxws/node` is the subpath that owns the `ws` socket, and it
* is the only place in the package that touches `ws`. A browser dialer imports `connect` from
* `muxws` instead and changes nothing else.
*/
import 'muxws';
import { connect } from 'muxws/node';
const url = process.env.MUXWS_URL ?? 'ws://127.0.0.1:8000/ws';
interface Greeting {
greeting: string;
}
interface Chunk {
chunk: number;
of: number;
}
const peer = await connect(url);
try {
// Unary. `request()` sends one payload, waits for exactly one back, and rejects if the remote
// sends a second - which `await stream` deliberately does not.
const answer = await peer.request<Greeting>({ say: 'hello' });
console.log(`greeting: ${answer.greeting}`);
// Streaming response. `open()` is synchronous: it hands back the Stream in the same turn it
// allocated the id. `end: true` says this side has nothing more to send, so the acceptor is free
// to stream back immediately.
const stream = peer.open<Chunk>({ count: 3 }, { end: true });
for await (const chunk of stream) {
console.log(`chunk ${chunk.chunk} of ${chunk.of}`);
}
} finally {
await peer.close();
}Run it in a second terminal:
npx tsx quickstart-client.tsSet MUXWS_URL to point either client somewhere other than ws://127.0.0.1:8000/ws.
4. What you see
Either client prints exactly this, and the documentation test asserts it byte for byte against both:
greeting: hello, muxws
chunk 1 of 3
chunk 2 of 3
chunk 3 of 3The first line came back on one stream and the last three on another, both over the same socket — no second connection, and no correlation id you had to invent. This client finishes the request before it opens the second stream, but nothing requires that: had both been open at once, the writer would have interleaved their frames, because a stream holds at most one unsent fragment at a time and the writer takes the streams round-robin. That is what stops a 1 MB export from stalling a 200-byte progress update on another stream.
Next: server push
There is no push API, because there does not need to be one. Peer is a single symmetric type: the acceptor calls peer.open() exactly as the dialer did, on the socket that is already there, and the dialer receives it in its on_stream handler. Same correlation, same cancellation, same everything.
The acceptor side — peer is in scope in the route, so the handler closes over it:
"""Server push: the acceptor opens a stream of its own.
Run it with `python push_server.py`; it listens on 127.0.0.1:8001 unless `MUXWS_PORT` says
otherwise, and `push_client.py` (or `push-client.ts`) is the other half.
There is no separate push API. `peer.open()` here is the same call the dialer makes in
`quickstart_client.py`, on the same socket, with the same correlation and the same cancellation -
which is the whole of muxws's symmetry claim.
"""
import os
from fastapi import FastAPI, WebSocket
from muxws import accept, Stream
HOST = "127.0.0.1"
PORT = int(os.environ.get("MUXWS_PORT", "8001"))
app = FastAPI()
@app.websocket("/ws")
async def muxws_endpoint(websocket: WebSocket) -> None:
peer = await accept(websocket)
async def on_stream(payload: object, _stream: Stream) -> None:
"""The dialer's one-shot notify. Nothing to answer here; the push is the interesting half."""
topic = payload["subscribe"] if isinstance(payload, dict) else "ticks"
# Opened by the acceptor, on its own initiative. The dialer sees it in *its* on_stream
# handler, exactly as this peer saw the dialer's stream.
ticks = peer.open({"topic": topic})
for n in (1, 2, 3):
await ticks.send({"tick": n})
await ticks.end()
peer.on_stream(on_stream)
await peer.serve()
if __name__ == "__main__":
import uvicorn
uvicorn.run(app, host=HOST, port=PORT, log_level="warning")The dialer side. The handler is passed to connect() rather than registered after it returns, because the acceptor may push a stream the instant it sees this dialer, and a handler registered one await later would meet that push with reset(REFUSED):
"""Server push, dialer half: a dialer with an `on_stream` handler of its own.
Start `push_server.py` first, then run `python push_client.py`. The URL is `ws://127.0.0.1:8001/ws`
unless `MUXWS_URL` says otherwise.
The handler is passed to `connect()` rather than registered after it returns, because the acceptor
may push a stream the instant it sees this dialer: a handler registered one `await` later would meet
that push with `reset(REFUSED)`.
"""
import asyncio
import os
from muxws import connect, Stream
URL = os.environ.get("MUXWS_URL", "ws://127.0.0.1:8001/ws")
async def main() -> None:
finished = asyncio.Event()
async def on_push(payload: object, stream: Stream) -> None:
print(f"push: {payload['topic']}")
async for tick in stream:
print(f"tick {tick['tick']}")
finished.set()
peer = await connect(URL, on_stream=on_push)
try:
# One-shot: `notify()` sends a payload and ends the stream, and hands back no handle to
# await, because there is nothing coming back on it.
await peer.notify({"subscribe": "ticks"})
# 10.0 seconds, and only so a stuck example fails rather than hangs.
await asyncio.wait_for(finished.wait(), timeout=10.0)
finally:
await peer.close()
if __name__ == "__main__":
asyncio.run(main())Or the same dialer in TypeScript:
/**
* Server push, dialer half, in TypeScript on Node - the other end of `push_server.py`.
*
* Run it with `npx tsx push-client.ts`. The URL is `ws://127.0.0.1:8001/ws` unless `MUXWS_URL` says
* otherwise.
*
* `onStream` is passed to `connect()` rather than registered after it returns, because the acceptor
* may push a stream the instant it sees this dialer: a handler registered one `await` later would
* meet that push with `reset(REFUSED)`.
*/
import 'muxws';
import { connect } from 'muxws/node';
const url = process.env.MUXWS_URL ?? 'ws://127.0.0.1:8001/ws';
interface Topic {
topic: string;
}
interface Tick {
tick: number;
}
let finished: () => void = () => undefined;
const pushDone = new Promise<void>((resolve) => {
finished = resolve;
});
const peer = await connect(url, {
onStream: async (payload, stream) => {
console.log(`push: ${(payload as Topic).topic}`);
for await (const tick of stream) {
console.log(`tick ${(tick as Tick).tick}`);
}
finished();
},
});
try {
// One-shot: `notify()` sends a payload and ends the stream, and hands back no handle to await,
// because there is nothing coming back on it.
await peer.notify({ subscribe: 'ticks' });
// 10000 milliseconds, and only so a stuck example fails rather than hangs.
const deadline = new Promise<never>((_resolve, reject) => {
setTimeout(() => {
reject(new Error('the push never arrived within 10000 milliseconds'));
}, 10_000).unref();
});
await Promise.race([pushDone, deadline]);
} finally {
await peer.close();
}Start python push_server.py (127.0.0.1:8001) and run either client. Both print:
push: ticks
tick 1
tick 2
tick 3Next: surviving a disconnect
Reconnection is a dialer-only helper, and you get it by passing two more arguments to connect(): a hello and a Reconnect.
"""Surviving a disconnect: the same dialer with `hello=` and `reconnect=Reconnect()`.
Start `quickstart_server.py` first, then run `python reconnect_client.py`. This asks once and exits,
so `on_reconnect` does not fire on a healthy socket: it is here to show where a reconnection would
surface in a program that stays connected.
"""
import asyncio
import os
from muxws import connect, Peer, Reconnect
URL = os.environ.get("MUXWS_URL", "ws://127.0.0.1:8000/ws")
def on_reconnect(reconnection: int, _peer: Peer) -> None:
"""Fires once per **re-established** connection: socket open, subprotocol accepted, hello answered.
The number is how many times this peer has reconnected - the first reconnection is 1 - and not
how many dial attempts that reconnection cost.
"""
print(f"reconnected (reconnection {reconnection})")
async def main() -> None:
peer = await connect(
URL,
# Captured once, here, by value, and replayed verbatim on every later connection. It is an
# ordinary stream: the acceptor sees it in its own on_stream handler and acknowledges it by
# returning. Never put a credential in it - authentication belongs to the upgrade.
hello={"client": "quickstart"},
# initial_delay 0.25 seconds, factor 2.0, max_delay 30.0 seconds, jitter 0.3, unlimited
# attempts. Every duration in the Python port is seconds as a float.
reconnect=Reconnect(),
on_reconnect=on_reconnect,
)
try:
answer = await peer.request({"say": "hello"})
print(f"greeting: {answer['greeting']}")
finally:
await peer.close()
if __name__ == "__main__":
asyncio.run(main())What that buys you: when an established connection is lost, the dialer redials on a jittered exponential schedule — initial_delay 0.25 seconds, doubling, capped at max_delay 30.0 seconds, ±30 % jitter — and replays the hello verbatim on every socket it ever gets. The acceptor sees the hello as an ordinary stream in its ordinary on_stream handler and acknowledges it by returning, so your identity is re-established without any code on the accepting side.
What it does not buy you: no stream survives. Every stream that was open when the socket died is already dead, nothing was buffered while the peer was between sockets, the stream id space starts over at 1, and peer.id changes — a reconnected peer reads as a new connection in the log, on purpose. A reconnect restores a live socket and an accepted identity, and that is the whole list.
One thing it deliberately does not do either: if the first connect() fails, it raises, whatever reconnect= says. A typo in the URL that retried forever would never surface.
Run it against the same quickstart_server.py. It connects, asks once and exits, so on a healthy socket the on_reconnect line never appears — it is in the file to show you where a reconnection surfaces, not because a two-second run will see one:
greeting: hello, muxws