WebSockets

WebSockets provide a full-duplex communication channel between the client and the server over a single long-lived connection, allowing both sides to send messages at any time. HTTPX has native support for them through client.websocket().

WebSocket support requires the optional wsproto dependency:

pip install 'httpx2[ws]'

Opening a session

client.websocket() is a context manager that performs the WebSocket handshake and yields a WebSocketSession:

>>> with httpx2.Client() as client:
...     with client.websocket("wss://example.com/ws") as ws:
...         ws.send_text("Hello!")
...         message = ws.receive_text()

It works the same way with the async client, using async with and await:

>>> async with httpx2.AsyncClient() as client:
...     async with client.websocket("wss://example.com/ws") as ws:
...         await ws.send_text("Hello!")
...         message = await ws.receive_text()

Both ws:// and wss:// URLs are supported, as well as their http:// and https:// equivalents. The handshake is a regular HTTP request, so the client's configuration - headers, cookies, authentication, proxies, and timeouts - applies as usual.

The session is closed automatically when exiting the context manager.

For quick one-off sessions, there is also a top-level httpx2.websocket() function, mirroring httpx2.request():

>>> with httpx2.websocket("wss://example.com/ws") as ws:
...     ws.send_text("Hello!")

Sending and receiving messages

The session provides send and receive methods for text, bytes, and JSON messages:

>>> ws.send_text("Hello!")
>>> ws.send_bytes(b"Hello!")
>>> ws.send_json({"message": "Hello!"})

>>> ws.receive_text()
'Hello!'
>>> ws.receive_bytes()
b'Hello!'
>>> ws.receive_json()
{'message': 'Hello!'}

JSON messages are sent as text frames by default. Pass mode="binary" to send_json() or receive_json() to use binary frames instead.

The receive methods block until a message is available. Pass a timeout in seconds to bound the wait:

>>> ws.receive_text(timeout=2.0)

If no message arrives in time, TimeoutError is raised.

If the received message doesn't match the expected type, WebSocketInvalidTypeReceived is raised.

Subprotocols

To negotiate a subprotocol with the server, pass the subprotocols argument. The subprotocol accepted by the server is available as ws.subprotocol:

>>> with client.websocket("wss://example.com/ws", subprotocols=["graphql-ws"]) as ws:
...     ws.subprotocol
'graphql-ws'

Keepalive pings

The session automatically sends a Ping event at a regular interval to keep the connection alive. If the server doesn't answer in time, the connection is considered lost and WebSocketNetworkError is raised.

You can also send a Ping manually. ping() returns an event that is set when the corresponding Pong is received:

>>> pong = ws.ping()
>>> pong.wait()  # Blocks until the server answers.

Configuration

client.websocket() and httpx2.websocket() accept the following keyword arguments to tune the session behaviour:

Accessing the handshake response

The Response that upgraded the connection is available as ws.response, which is useful for inspecting the handshake headers:

>>> with client.websocket("wss://example.com/ws") as ws:
...     ws.response.headers

Error handling

WebSocket exceptions live in the httpx2.websockets namespace:

>>> from httpx2.websockets import WebSocketDisconnect

• WebSocketUpgradeError - the server didn't accept the upgrade, and responded with something other than a 101 Switching Protocols status code. The response is available as .response.
• WebSocketDisconnect - the server closed the session. The close code and reason are available as .code and .reason.
• WebSocketInvalidTypeReceived - the received message didn't match the expected type.
• WebSocketNetworkError - a network error occurred, typically because the underlying stream closed or timed out.

All of these are subclasses of httpx2.websockets.HTTPXWSException:

>>> from httpx2.websockets import WebSocketDisconnect
>>> with client.websocket("wss://example.com/ws") as ws:
...     try:
...         message = ws.receive_text()
...     except WebSocketDisconnect as exc:
...         print(f"Connection closed: {exc.code} {exc.reason}")

Testing ASGI applications

Just as ASGITransport lets you make HTTP requests directly against an ASGI application, ASGIWebSocketTransport extends it with WebSocket support - useful for testing Starlette or FastAPI applications without running a server:

from httpx2 import AsyncClient
from httpx2.websockets import ASGIWebSocketTransport

async with AsyncClient(transport=ASGIWebSocketTransport(app)) as client:
    async with client.websocket("ws://testserver/ws") as ws:
        await ws.send_text("Hello!")
        message = await ws.receive_text()

Regular HTTP requests through the same transport work too, so a single client can exercise the whole application.

Acknowledgements

WebSocket support is derived from the excellent httpx-ws package by François Voron.