httpx-ws

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())