HTTPX WebSockets

0.9.0 · active · verified Thu Apr 09

httpx-ws is a Python library that adds WebSocket client capabilities to the popular HTTPX library. It provides both synchronous and asynchronous APIs for connecting to WebSocket servers and sending/receiving messages, seamlessly integrating with HTTPX's client infrastructure. Currently at version 0.9.0, it maintains an active release cycle with frequent updates and improvements.

Warnings

breaking Async exceptions, including `WebSocketDisconnect`, are now wrapped in `ExceptionGroup` when propagating out of the `async with` block for `WebSocketClient` (or `aconnect_ws`).
Fix: Update `try...except` blocks to use the `except*` syntax for specific sub-exceptions (e.g., `except* WebSocketDisconnect`) or catch `ExceptionGroup` and iterate its `exceptions` attribute.
breaking Python 3.9 support was dropped in v0.8.0, and Python 3.8 support was dropped in v0.7.0. The library now requires Python 3.10 or newer.
Fix: Ensure your project is running on Python 3.10 or a newer compatible version.
gotcha A new class-based API, `WebSocketClient`, was introduced in v0.8.0 and is now the recommended way to open WebSocket connections, replacing the older `connect_ws`/`aconnect_ws` functions.
Fix: Migrate from `connect_ws`/`aconnect_ws` functions to `WebSocketClient` for new code and consider refactoring existing code to use the more flexible class-based approach (e.g., `async with httpx.AsyncClient() as client: async with WebSocketClient(url, client) as ws:`).
breaking The `subprotocol` parameter was removed from `AsyncWebSocketSession` and `WebSocketSession` constructors. Subprotocols are now automatically set from response headers.
Fix: If you were directly instantiating `AsyncWebSocketSession` or `WebSocketSession` and passing `subprotocol`, remove this parameter. If using `connect_ws`/`aconnect_ws` or `WebSocketClient`, this change is typically transparent.

Install

pip install httpx-ws Install stable version

Imports

WebSocketClient
```
from httpx_ws import WebSocketClient
```
Recommended class-based client API introduced in v0.8.0.
aconnect_ws
```
from httpx_ws import aconnect_ws
```
Older function-based async connection API; the class-based client is now preferred.
connect_ws
```
from httpx_ws import connect_ws
```
Older function-based sync connection API; the class-based client is now preferred.

WebSocketDisconnect

from httpx_ws import WebSocketDisconnect

Quickstart

Demonstrates connecting to a WebSocket server using the class-based `WebSocketClient`, sending text, and receiving text. It includes robust error handling for `WebSocketDisconnect` and `ExceptionGroup` (for v0.9.0+). Replace `ws://localhost:8000/ws` with your target WebSocket endpoint. Requires an active WebSocket server running at the specified URL to execute successfully.

import httpx
from httpx_ws import WebSocketClient, WebSocketDisconnect
import asyncio

async def main():
    # Connect to a local test WebSocket server (e.g., uvicorn with websockets)
    # For a real application, replace with your actual WebSocket URL
    websocket_url = "ws://localhost:8000/ws"
    
    try:
        async with httpx.AsyncClient() as client:
            async with WebSocketClient(websocket_url, client) as websocket:
                print(f"Connected to {websocket_url}")
                await websocket.send_text("Hello, WebSocket!")
                print("Sent: Hello, WebSocket!")
                
                # Try to receive a response, with a timeout
                try:
                    data = await asyncio.wait_for(websocket.receive_text(), timeout=5.0)
                    print(f"Received: {data}")
                except asyncio.TimeoutError:
                    print("No data received within 5 seconds.")
                
    except* WebSocketDisconnect as e:
        print(f"WebSocket disconnected gracefully: {e}")
    except ExceptionGroup as eg:
        # Handle other exceptions wrapped in ExceptionGroup (v0.9.0+)
        # This ensures all potential wrapped errors are considered
        print(f"An ExceptionGroup occurred: {eg}")
        for exc in eg.exceptions:
            if isinstance(exc, WebSocketDisconnect):
                print(f"Caught WebSocketDisconnect within ExceptionGroup: {exc}")
            else:
                print(f"Caught other exception within ExceptionGroup: {exc}")
                raise exc # Re-raise if not specifically handled
    except Exception as e:
        print(f"An unexpected error occurred: {e}")

if __name__ == "__main__":
    asyncio.run(main())

view raw JSON →