mrkeyoor.com_
Tue 22 Sept 01:45 UTC
PyPIWeb Backendupdated 21 Sept 2026

trio-websocket review

trio-websocket connects wsproto's WebSocket state machine to Trio streams, nurseries, and cancellation. It provides client context managers, a server handshake object, message send and receive calls, ping/pong handling, close codes, subprotocol negotiation, TLS, queue limits, and bounded message size. The project reports passing the Autobahn protocol suite. Version 0.12.2 fixes the port chosen for a `wss://` URL when no explicit SSL context is supplied. Our Python 3.12 install found a typed, pure-Python package whose import took 0.51 seconds.

Verdict

trio-websocket is still the direct WebSocket choice for a Trio-native service that needs handshake and protocol control. Do not adopt it for asyncio or framework-level HTTP routing, and budget for life-support maintenance rather than a busy roadmap.

We installed it

Lab card: what happened when we installed trio-websocketScreenshot of trio-websocket documentation
Install✓ · 0.5s9 packages on disk · 4 MB
Importimport trio_websocket in 0.51s · pure Python · py.typed · requires Python >=3.8
Known vulns0(pip-audit)

Answers from our run

Does trio-websocket install cleanly?

Yes. In a fresh container with an empty cache, pip install trio-websocket finished in 0.5s, leaving 9 packages and 4 MB on disk. pip-audit reported no known vulnerabilities.

What does trio-websocket need to run?

Python >=3.8, and nothing compiled: it is pure Python. In our run import trio_websocket succeeded in 0.51s, and the package ships py.typed for type checkers.

trio-websocket or websockets: which should you use?

websockets: Choose it for an asyncio client or server, a sync API, and a larger direct-user community. trio-websocket is still the direct WebSocket choice for a Trio-native service that needs handshake and protocol control.

When should you not use trio-websocket?

The application runs on asyncio. trio-websocket has no asyncio backend; the websockets package fits that event loop directly.

API stability4/5Client and server entry points, request.accept(), request.reject(), send_message(), get_message(), ping(), and aclose() have stayed recognizable across recent 0.x releases. Version 0.12 added strict ExceptionGroup support, more annotations, and receive-buffer configuration, while 0.12.2 fixed secure default-port selection. The behavior is steady enough for Selenium and Trio applications, but pre-1.0 status and exception wrapping still justify pinned versions and integration tests.
Docs4/5Read the Docs has client and server walkthroughs plus parameter-level references for URLs, headers, TLS, subprotocols, message limits, queue sizing, pings, close reasons, custom streams, listeners, and exceptions. The README gives working minimal examples and states the maintenance status plainly. Less common production questions, including ExceptionGroup boundaries, multi-reader behavior, proxy integration, and server shutdown structure, require careful reference reading or source inspection.
Maintenance2/5The repository is unarchived, but its own README calls the project life-support maintenance. PyPI uploaded 0.12.2 in February 2025, GitHub shows the last push in October 2025, and the repository has 80 stars with 21 open issues and pull requests. The 0.12 series fixed dependency declaration, exception context, Trio compatibility, and secure URL port handling, so critical upkeep happens, while new feature capacity is explicitly limited.
Ecosystem3/5The provided latest-week count is 10,621,898 downloads, driven heavily by Selenium's Python dependency chain. Trio, wsproto, Quart Trio, and HTTPX WS provide a coherent set of neighboring tools, and the Autobahn suite gives protocol confidence. Direct community reach is much smaller than the download number suggests: 80 stars, life-support status, and no asyncio mode mean most WebSocket tutorials and integrations target other libraries.

Use it if

  • The application already uses Trio and wants WebSocket connection lifetime tied to cancel scopes and nurseries.
  • A server must inspect headers and path, choose a subprotocol, or reject the HTTP upgrade before accepting a socket.
  • Message-size, receive-buffer, queue, connect, and disconnect limits need direct configuration at the protocol boundary.
  • You maintain Selenium or another dependency that already brings trio-websocket into the environment.
Skip it if

Setup reality

We installed trio-websocket 0.12.2 in a fresh Python 3.12 Bookworm container. Installation finished in 0.5 seconds, left 9 packages, and used 4 MB. It declares 4 direct dependencies, requires Python >=3.8, and is pure Python under the MIT License. pip-audit found no known vulnerabilities. The wheel includes py.typed, and import trio_websocket succeeded in 0.51 seconds. There are no native builds, credentials, or config files.

A plain server call still needs ssl_context=None; omitting the argument raises TypeError. For wss:// clients, 0.12.2 fixes the default secure port when no context is passed. A custom client context supplies private CAs or TLS policy. The URL scheme decides whether transport is encrypted. Client headers are byte pairs, while server request headers also arrive as byte pairs rather than a case-insensitive mapping.

serve_websocket() runs until cancellation. Start it through nursery.start() when code needs the bound port or a clean test lifecycle. Default message size is 1 MiB, message queue size is 1, receive buffer is 4 KiB, and connect and disconnect timeouts are each 60 seconds. A larger message closes with code 1009. Raising the queue reduces immediate backpressure and consumes more memory per busy connection.

Trio now defaults to strict exception groups, and trio-websocket 0.12 supports them. An error originating in the internal nursery can arrive inside an ExceptionGroup, so except* or exceptiongroup.catch() may be required around connection scopes. Multiple tasks may send because a FIFO lock serializes writes. Multiple readers split messages unpredictably, so keep one get_message() consumer and distribute decoded events yourself. Wrap idle receives and ping waits in Trio timeouts because get_message() has no built-in idle deadline.

Patterns

Open a client connection connect-client

import trio
from trio_websocket import open_websocket_url

async def main():
    async with open_websocket_url('wss://example.com/socket') as socket:
        await socket.send_message('hello')
        print(await socket.get_message())

trio.run(main)

The context manager performs connect and close handshakes. TCP errors and rejected upgrades are raised when entering it.

Serve a WebSocket echo endpoint serve-echo

import trio
from trio_websocket import ConnectionClosed, serve_websocket

async def handler(request):
    socket = await request.accept()
    try:
        while True:
            message = await socket.get_message()
            await socket.send_message(message)
    except ConnectionClosed:
        pass

trio.run(serve_websocket, handler, '127.0.0.1', 8000, None)

The final positional None is the required ssl_context for a plaintext ws server.

Read a kernel-assigned server port start-test-server

from functools import partial

async with trio.open_nursery() as nursery:
    server = await nursery.start(
        partial(serve_websocket, handler, '127.0.0.1', 0, ssl_context=None)
    )
    await run_checks(server.port)
    nursery.cancel_scope.cancel()

nursery.start() returns only after listeners are ready, which removes connection races in tests.

Reject a handshake before accepting reject-upgrade

async def handler(request):
    headers = {name.lower(): value for name, value in request.headers}
    if headers.get(b'authorization') != expected_header:
        await request.reject(401, body=b'unauthorized\n')
        return
    socket = await request.accept()

Headers are byte tuples. Authenticate before accept() so rejected clients receive an HTTP response instead of an established WebSocket.

Negotiate a shared subprotocol choose-subprotocol

async def handler(request):
    supported = ['events.v2', 'events.v1']
    chosen = next(
        (item for item in supported if item in request.proposed_subprotocols),
        None,
    )
    if chosen is None:
        await request.reject(400, body=b'no supported subprotocol')
        return
    socket = await request.accept(subprotocol=chosen)

Only select a value proposed by the client. After acceptance, both ends can read the selected subprotocol.

Inspect the peer's close reason read-close-reason

from trio_websocket import ConnectionClosed

try:
    while True:
        handle(await socket.get_message())
except ConnectionClosed as error:
    if error.reason is not None:
        log.info('closed code=%s reason=%s', error.reason.code, error.reason.reason)

Queued messages are delivered before a remote-close exception once the receive queue drains.

Run one reader and one writer send-receive-concurrently

async def reader(socket):
    while True:
        handle(await socket.get_message())

async def writer(socket, outbound):
    async for message in outbound:
        await socket.send_message(message)

async with trio.open_nursery() as nursery:
    nursery.start_soon(reader, socket)
    nursery.start_soon(writer, socket, outbound)

Writes are serialized internally. Keep a single reader, then fan messages out through application channels if several consumers need them.

Stop waiting on an idle connection bound-idle-wait

with trio.move_on_after(30) as scope:
    message = await socket.get_message()

if scope.cancelled_caught:
    await socket.aclose(code=1001, reason='idle timeout')

get_message() has no idle timeout. Trio cancel scopes provide the deadline.

Set server memory and message bounds configure-limits

await serve_websocket(
    handler,
    '0.0.0.0',
    8000,
    ssl_context=None,
    max_message_size=256 * 1024,
    message_queue_size=8,
    receive_buffer_size=16 * 1024,
)

Messages over max_message_size close with code 1009. Queue and buffer values multiply across concurrent connections.

Put a deadline around ping check-liveness

with trio.move_on_after(10) as scope:
    await socket.ping()

if scope.cancelled_caught:
    await socket.aclose(code=1011, reason='ping timeout')

ping() waits for its matching pong. Without a deadline, a dead peer can leave the task waiting for the TCP stack.

Serve secure WebSockets serve-tls

import ssl

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

await serve_websocket(
    handler, '0.0.0.0', 8443, ssl_context=context
)

Use wss URLs for clients. Version 0.12.2 corrects the implicit port when a wss URL is opened without a supplied context.

Handle grouped connection failures catch-exception-group

from trio_websocket import ConnectionRejected, HandshakeError

try:
    async with open_websocket_url(url) as socket:
        await socket.send_message('hello')
except* ConnectionRejected as group:
    for error in group.exceptions:
        log.warning('upgrade rejected: %s', error.status_code)
except* HandshakeError as group:
    log.error('handshake failed: %r', group.exceptions)

Strict exception groups can wrap failures from internal nurseries. Python 3.10 code can use the exceptiongroup backport's catch() helper.

Alternatives

PackageRegistryPick it when
websocketsPyPIChoose it for an asyncio client or server, a sync API, and a larger direct-user community.
quart-trioPyPIChoose it when Trio HTTP routes and WebSocket endpoints belong in one framework application.
httpx-wsPyPIChoose it when WebSockets should reuse HTTPX authentication, cookies, and transport configuration.
wsprotoPyPIChoose it when you need only the sans-I/O protocol engine and will supply all networking and task management.

More web backend guides

urllib3 · requests · ws · anyio · httpx · undici · the whole shelf →

How this guide is made: grounded in the library's documentation, release notes, changelog, and issue history, on a fixed rubric — not a hands-on install of every release. The 50 most-downloaded entries are additionally install-verified in clean containers. Corrections: contact the desk.