PyPIWeb Backendupdated 07 Aug 2026

trio-websocket

Item: trio-websocket
Rating: 3.3
Author: Keyoor

trio-websocket is a WebSocket client and server for Trio. It pairs wsproto, a sans-IO state machine that implements RFC 6455 framing and handshakes, with Trio's networking and nursery model, so a connection is an async context manager whose lifetime belongs to a scope instead of an object you have to remember to close. It passes the Autobahn test suite, enforces a maximum message size, and performs a real closing handshake with a timeout rather than dropping the socket. Most of its download volume is not people choosing it: Selenium's Python bindings require trio-websocket for their browser WebSocket transport, so it lands in every environment that installs Selenium.

trio-websocket · v0.12.2 · 9.6M/wk · ★ 80 · pushed 21 Oct 2025 · python-trio/trio-websocket · docs · reviewed by Keyoor

Verdict

If your application runs on Trio and needs WebSockets, this is still the right answer and the protocol handling is careful. Go in knowing the maintainers describe it as life-support maintenance, so budget time for reading the source when the docs stop short.

API stability4/5The surface has barely moved since 0.10 and Selenium pins to it, which keeps it honest. It is still 0.x and labelled alpha on PyPI, and 0.12 did change how exceptions reach you under trio's strict exception groups.

Docs4/5trio-websocket.readthedocs.io has a real narrative section plus a full API reference, and the source docstrings document every parameter and every exception each call can raise. Less common paths, such as wrapping your own stream or custom listeners, are reference-only.

Maintenance2/5The README itself says the project is on life-support maintenance and asks for help. Last release February 2025, last commit October 2025, 17 open issues out of 21 open issues and PRs, and PyPI still says Development Status 3 - Alpha after seven years.

Ecosystem3/5Roughly 9.58M weekly downloads against 80 stars, because Selenium requires it rather than because people picked it, so there are few blog posts, no plugins and little third-party code to copy. The Selenium pin does mean breakage on a new Python gets noticed quickly.

Use it if

Your application already runs on Trio and you want connection lifetime bound to a nursery and cancel scope instead of tasks you track by hand
You need the server side with real control over the handshake: read the request path and headers, pick a subprotocol, or reject the upgrade with a specific HTTP status code and response body
Protocol correctness matters to you: it passes the Autobahn suite, closes with code 1009 when a peer exceeds max_message_size, and exposes the close code and reason on ConnectionClosed
You are debugging or extending something that already depends on it, most commonly Selenium, and you need to know how the connection actually behaves

Skip it if

You are on asyncio. There is no asyncio backend and no AnyIO compatibility layer; this is Trio or nothing, and the websockets library is the mainstream choice everywhere else
You want an actively developed dependency: the README states the project is on life-support maintenance and asks for contributors, PyPI still classifies it Development Status 3 - Alpha, the last release 0.12.2 is from February 2025, and the last commit is October 2025 with 17 open issues out of 21 open issues and PRs
You only need a server and already have a web framework: the project's own README points at Quart through quart-trio, which serves HTTP and WebSocket from one app on one port
You want WebSocket alongside your HTTP client: this does not plug into httpx, so you end up with two connection stacks and two sets of timeout and proxy settings
You need a stable 1.0 contract: it is still 0.x, every downstream pin in the wild is <1.0, and 0.12 changed how errors surface once trio's strict_exception_groups became the default

Setup reality

pip install trio-websocket is pure Python with four dependencies (trio, wsproto, outcome, and exceptiongroup on Python 3.10 and below) and no compiler step. The friction comes from Trio's model, not the install. serve_websocket takes ssl_context as a required argument with no default, so plain ws:// means passing ssl_context=None explicitly, and forgetting it is the first error most people hit. serve_websocket also never returns, so if you want the bound port or an orderly shutdown you have to start it with nursery.start() and hold the returned WebSocketServer. Since 0.12 the library supports trio's strict_exception_groups=True, which is now trio's default, so a plain except HandshakeError around a connection may not match once the error arrives wrapped in an ExceptionGroup. Defaults worth knowing before you deploy: max_message_size is 1 MiB and a larger frame closes the connection with code 1009, message_queue_size is 1, and connect_timeout and disconnect_timeout are both 60 seconds.

Patterns

Open a client connection to a URLconnect-client-to-url

import trio
from trio_websocket import open_websocket_url

async def main():
    async with open_websocket_url('wss://echo.websocket.org') as ws:
        await ws.send_message('hello world!')
        message = await ws.get_message()
        print('Received:', message)

trio.run(main)

The context manager connects on entry and runs the closing handshake on exit, so there is no explicit close call. A refused TCP connection raises OSError and a failed upgrade raises HandshakeError, both out of the async with line.

Run a WebSocket serverserve-echo-server

import trio
from trio_websocket import ConnectionClosed, serve_websocket

async def echo_server(request):
    ws = await request.accept()
    while True:
        try:
            message = await ws.get_message()
            await ws.send_message(message)
        except ConnectionClosed:
            break

async def main():
    await serve_websocket(echo_server, '127.0.0.1', 8000, ssl_context=None)

trio.run(main)

ssl_context has no default value, so plain ws:// requires passing None explicitly or you get a TypeError. serve_websocket blocks forever; it only ends when the surrounding cancel scope is cancelled.

Start a server in a nursery and get its bound portstart-server-and-read-port

import trio
from functools import partial
from trio_websocket import serve_websocket

async def main():
    async with trio.open_nursery() as nursery:
        server = await nursery.start(
            partial(serve_websocket, handler, '127.0.0.1', 0, ssl_context=None)
        )
        print('listening on', server.port)
        await run_client_tests(server.port)
        nursery.cancel_scope.cancel()

trio.run(main)

nursery.start returns the WebSocketServer only once it is accepting connections, which is the only reliable way to learn a kernel-assigned port after passing 0. server.port raises RuntimeError when the server has more than one listener, which is what happens if you bind host=None.

Read the close code and reasonhandle-connection-closed

from trio_websocket import ConnectionClosed

async def drain(ws):
    while True:
        try:
            print(await ws.get_message())
        except ConnectionClosed as exc:
            reason = exc.reason  # CloseReason or None
            if reason is not None:
                print(reason.code, reason.name, reason.reason)
            break

Messages that arrived before the remote closed are still delivered; ConnectionClosed only fires once the queue drains. If your side closed first, pending messages are discarded and the next get_message raises immediately.

Authenticate before upgrading the connectionreject-handshake

async def handler(request):
    headers = {name.lower(): value for name, value in request.headers}
    if headers.get(b'authorization') != b'Bearer secret':
        await request.reject(401, body=b'unauthorized\n')
        return
    ws = await request.accept()
    await ws.send_message('welcome ' + request.path)

request.headers is a list of byte tuples, not a mapping, so normalise names yourself. reject takes the status code positionally with keyword-only extra_headers and body, and the code must not be 101.

Propose and select a subprotocolnegotiate-subprotocol

# client
async with open_websocket_url(
    'wss://example.com/ws',
    subprotocols=['v2.json', 'v1.json'],
    extra_headers=[(b'authorization', b'Bearer token')],
) as ws:
    print(ws.subprotocol)  # what the server chose, or None

# server
async def handler(request):
    chosen = 'v2.json' if 'v2.json' in request.proposed_subprotocols else None
    ws = await request.accept(subprotocol=chosen)

extra_headers must be bytes tuples, not strings. Only name a subprotocol in accept() that appears in request.proposed_subprotocols; when nothing is negotiated, ws.subprotocol is None on both sides.

Read and write from separate tasksconcurrent-send-and-receive

import trio
from trio_websocket import ConnectionClosed, open_websocket_url

async def reader(ws):
    while True:
        print('got', await ws.get_message())

async def writer(ws):
    while True:
        await ws.send_message('ping')
        await trio.sleep(5)

async def main():
    async with open_websocket_url('wss://example.com/ws') as ws:
        try:
            async with trio.open_nursery() as nursery:
                nursery.start_soon(reader, ws)
                nursery.start_soon(writer, ws)
        except* ConnectionClosed:
            print('peer went away')

trio.run(main)

Sends are serialized internally by a FIFO lock, so several tasks may call send_message safely. Do not call get_message from two tasks unless you want each message delivered to exactly one of them at random.

Bound connect, receive and close timesapply-timeouts

import trio
from trio_websocket import open_websocket_url

async def main():
    async with open_websocket_url(
        'wss://example.com/ws',
        connect_timeout=10,
        disconnect_timeout=5,
    ) as ws:
        with trio.move_on_after(30) as scope:
            message = await ws.get_message()
        if scope.cancelled_caught:
            print('no traffic for 30s')

trio.run(main)

connect_timeout and disconnect_timeout both default to 60 seconds and raise ConnectionTimeout and DisconnectionTimeout, which are subclasses of HandshakeError. There is no idle timeout on get_message, so wrap it in a Trio cancel scope yourself.

Cap message size and apply backpressurelimit-message-size-and-queue

from trio_websocket import serve_websocket

await serve_websocket(
    handler,
    '0.0.0.0',
    8000,
    ssl_context=None,
    max_message_size=256 * 1024,   # default is 1 MiB
    message_queue_size=16,         # default is 1
    receive_buffer_size=16 * 1024, # default is 4 KiB
)

A peer that exceeds max_message_size has its connection closed with code 1009 (Message Too Big); the message is never handed to your handler. message_queue_size is the backpressure knob: at the default of 1, a slow handler stops the reader from pulling more frames off the socket.

Check that the peer is aliveping-for-liveness

import trio
from trio_websocket import ConnectionClosed

async def keepalive(ws, interval=20):
    while True:
        await trio.sleep(interval)
        with trio.move_on_after(10) as scope:
            await ws.ping()
        if scope.cancelled_caught:
            await ws.aclose(code=1011, reason='ping timeout')
            return

ping() does not return until the matching pong arrives, so without a cancel scope a dead peer leaves that task hanging until the TCP layer notices. Passing the same payload while an earlier ping with that payload is still in flight raises ValueError; leave payload as None for a random one.

Serve and connect over wss with TLSserve-over-tls

import ssl
import trio
from trio_websocket import open_websocket_url, serve_websocket

server_ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER)
server_ctx.load_cert_chain('server.crt', 'server.key')

async def main():
    await serve_websocket(handler, '0.0.0.0', 8443, ssl_context=server_ctx)

# client against a private CA
client_ctx = ssl.create_default_context(cafile='ca.crt')
# async with open_websocket_url('wss://host:8443/', client_ctx) as ws: ...

For a client, ssl_context is the second positional argument and passing None means the default verifying context, not plaintext; the scheme in the URL decides ws or wss. A hostname mismatch surfaces as an ssl.SSLCertVerificationError out of the context manager, not as HandshakeError.

Catch handshake errors under strict exception groupscatch-errors-in-exception-groups

import trio
from trio_websocket import HandshakeError, ConnectionRejected, open_websocket_url

async def main():
    try:
        async with open_websocket_url('wss://example.com/ws') as ws:
            await ws.send_message('hi')
    except* ConnectionRejected as eg:
        for exc in eg.exceptions:
            print('rejected with', exc.status_code)
    except* HandshakeError as eg:
        print('handshake failed:', eg.exceptions)

trio.run(main)

Trio defaults to strict_exception_groups=True, so errors raised inside the connection's internal nursery reach you wrapped and a plain except HandshakeError can silently fail to match. On Python 3.10 use the exceptiongroup backport's catch() instead of except* syntax.

Alternatives

Package	Registry	Pick it when
websockets	PyPI	You are on asyncio, or you want the widely used client and server with a sync API option and far more community answers.
quart-trio	PyPI	You need HTTP routes and WebSocket endpoints in one Trio application served from a single port.
httpx-ws	PyPI	You want WebSocket support that reuses your httpx client, auth and transport, on either asyncio or Trio.
wsproto	PyPI	You want just the sans-IO protocol state machine and intend to drive the sockets yourself.