Trio WebSocket

raw JSON →
0.12.2 verified Tue May 12 auth: no python install: verified quickstart: stale maintenance

Trio WebSocket is a Python library that provides a WebSocket implementation for the Trio asynchronous I/O framework. It offers both client and server functionalities, built on the `wsproto` sans-IO state machine, emphasizing safety, correctness, and ergonomics. The current stable version is 0.12.2. While mature and stable, the project is currently in 'life-support maintenance', with contributions welcomed.

pip install trio-websocket

Common errors

error ConnectionRefusedError: [Errno 111] Connection refused ↓
cause The WebSocket client attempted to connect to a server that was not running or was not listening on the specified host and port.
fix
Ensure the WebSocket server is running and accessible on the target host and port before the client attempts to connect.
error trio_websocket.HandshakeError: expected status 101, got 404 ↓
cause The server responded with an HTTP status code other than 101 (Switching Protocols), indicating that it did not complete the WebSocket handshake successfully.
fix
Verify the WebSocket server's endpoint URL is correct and that the server implementation properly handles the WebSocket upgrade request.
error TypeError: object <coroutine object WebSocketConnection.send_message at 0x...> is not awaitable ↓
cause An asynchronous method of the `WebSocketConnection` object, such as `send_message` or `get_message`, was called without the `await` keyword.
fix
Prefix the call to the asynchronous method with await. For example, await ws.send_message(message).
error RuntimeError: must be run in a Trio task ↓
cause A Trio-specific function, such as `trio_websocket.connect_websocket` or `serve_websocket`, was called directly without being inside an `async def main():` function invoked by `trio.run(main)`.
fix
Wrap your application's entry point in an async def main(): function and then call trio.run(main()) to execute it within the Trio event loop.

Warnings

gotcha The `trio-websocket` project is currently in 'life-support maintenance'. While stable, active development and new features may be slow, and future major changes might occur with limited dedicated support. ↓
fix Be aware of the project's maintenance status. For critical applications, consider contributing or assessing alternatives if more active development is required. The `trio` framework itself is actively maintained.
gotcha When sending data, be mindful of the type (string or bytes). If a server is configured with `encoding="text"` but receives raw bytes, it might process them unexpectedly (e.g., as `None` or an empty string), even though `send_message` theoretically accepts both. ↓
fix Ensure that the data type sent by the client (`str` or `bytes`) matches the expected encoding/decoding on the server side. Explicitly encode/decode if necessary, especially when interoperating with non-Python WebSocket implementations.
gotcha Unlike some other async libraries where `BrokenStreamError` might be raised on a closed connection, `trio-websocket` raises `ConnectionClosed` when attempting to write to a connection that has already been closed. ↓
fix Handle `ConnectionClosed` specifically when expecting a client or server connection to gracefully shut down or when an attempt to send data might occur after closure. This allows for more precise error handling related to WebSocket lifecycle events.
gotcha Trio generally encourages caller-enforced timeouts using cancel scopes (`trio.fail_after`). However, `trio-websocket`'s high-level APIs (like `open_websocket_url` and server setup) incorporate internal timeouts for safety and ergonomics. This can be a point of confusion for users accustomed to Trio's 'timeout for humans' philosophy. ↓
fix Understand that high-level `trio-websocket` functions have built-in timeouts. If finer-grained control is needed, these can often be disabled or adjusted, and `trio.fail_after` can be wrapped around `trio-websocket` operations for explicit, composable timeout management.

Install compatibility verified last tested: 2026-05-12

python os / libc status wheel install import disk
3.10 alpine (musl) wheel - 0.44s 24.4M
3.10 alpine (musl) - - 0.47s 24.4M
3.10 slim (glibc) wheel 2.2s 0.31s 25M
3.10 slim (glibc) - - 0.35s 25M
3.11 alpine (musl) wheel - 0.55s 27.0M
3.11 alpine (musl) - - 0.62s 27.0M
3.11 slim (glibc) wheel 2.5s 0.49s 28M
3.11 slim (glibc) - - 0.48s 28M
3.12 alpine (musl) wheel - 0.51s 18.6M
3.12 alpine (musl) - - 0.49s 18.6M
3.12 slim (glibc) wheel 2.3s 0.48s 19M
3.12 slim (glibc) - - 0.52s 19M
3.13 alpine (musl) wheel - 0.46s 18.3M
3.13 alpine (musl) - - 0.47s 18.2M
3.13 slim (glibc) wheel 2.3s 0.46s 19M
3.13 slim (glibc) - - 0.52s 19M
3.9 alpine (musl) wheel - 0.42s 23.7M
3.9 alpine (musl) - - 0.76s 23.7M
3.9 slim (glibc) wheel 2.6s 0.38s 24M
3.9 slim (glibc) - - 0.39s 24M

Imports

Quickstart stale last tested: 2026-04-24

This quickstart demonstrates a minimal `trio-websocket` client and server. The server, `echo_server_handler`, binds to `127.0.0.1:8000` and reverses any text messages it receives before sending them back. The client, `client_task`, connects to this server, sends a test message, and asserts that it receives the reversed message back. Both are run concurrently within a Trio nursery.

import trio
import logging
from trio_websocket import serve_websocket, open_websocket_url, ConnectionClosed, WebSocketRequest

logging.basicConfig(level=logging.INFO)

async def echo_server_handler(request: WebSocketRequest) -> None:
    """Reverses incoming websocket messages and sends them back."""
    logging.info('Server: Handler starting on path "%s"', request.path)
    ws = await request.accept()
    try:
        while True:
            message = await ws.get_message()
            logging.info('Server: Received message: %s', message)
            await ws.send_message(message[::-1]) # Echo reversed message
    except ConnectionClosed:
        logging.info('Server: Connection closed.')
    except Exception as e:
        logging.error('Server: Error in handler: %s', e)
    finally:
        logging.info('Server: Handler exiting.')

async def client_task(port: int) -> None:
    """Connects to the server, sends a message, and waits for a response."""
    uri = f'ws://127.0.0.1:{port}/echo'
    logging.info('Client: Connecting to %s', uri)
    try:
        async with open_websocket_url(uri) as ws:
            test_message = 'hello trio-websocket'
            logging.info('Client: Sending: "%s"', test_message)
            await ws.send_message(test_message)

            received_message = await ws.get_message()
            expected_message = test_message[::-1] # Reversed
            logging.info('Client: Received: "%s"', received_message)

            assert received_message == expected_message
            logging.info('Client: Message echoed and reversed successfully!')
    except OSError as ose:
        logging.error('Client: Connection attempt failed: %s', ose)
    except Exception as e:
        logging.error('Client: Error in client task: %s', e)
    finally:
        logging.info('Client: Task finished.')

async def main():
    server_port = 8000
    async with trio.open_nursery() as nursery:
        logging.info(f'Main: Spawning server on 127.0.0.1:{server_port}')
        nursery.start_soon(serve_websocket, echo_server_handler, '127.0.0.1', server_port, None)
        
        # Give the server a moment to start up
        await trio.sleep(0.1)
        
        logging.info('Main: Spawning client task.')
        nursery.start_soon(client_task, server_port)

if __name__ == '__main__':
    try:
        trio.run(main)
    except KeyboardInterrupt:
        logging.info('Program interrupted by user.')
last verified Tue May 12
next check Fri Jun 26