websockets
WebSocket server and client library for Python — clean async API for both WebSocket server and client. websockets 12.x features: websockets.connect() for async client, websockets.serve() for server, send()/recv() for messages, ping/pong heartbeats, connection close handling, Unix domain socket support, SSL/TLS, subprotocol negotiation, compression (permessage-deflate), basic HTTP auth, broadcast() for sending to multiple connections, serve_forever() and connection event hooks.
Score Breakdown
⚙ Agent Friendliness
🔒 Security
WebSocket library. Always use wss:// (TLS) in production. Validate and authenticate connections on open. Rate limit incoming messages per connection. Validate message format before processing. Origin checking for browser clients. max_size parameter to limit message size and prevent memory exhaustion.
⚡ Reliability
Best When
Real-time streaming connections, agent-to-agent WebSocket communication, and bidirectional streaming — websockets is the cleanest pure-WebSocket async library.
Avoid When
REST APIs (use httpx/FastAPI), automatic reconnection needed (implement manually), or when FastAPI's WebSocket endpoint is already in the stack.
Use Cases
- • Agent WebSocket client — import asyncio, websockets; async def connect(): async with websockets.connect('wss://api.example.com/ws') as ws: await ws.send(json.dumps({'type': 'subscribe', 'channel': 'updates'})); async for msg in ws: data = json.loads(msg); handle(data) — client; agent maintains WebSocket connection to streaming API; async for receives messages
- • Agent WebSocket server — async def handler(websocket): async for message in websocket: data = json.loads(message); result = process(data); await websocket.send(json.dumps(result)); async with websockets.serve(handler, 'localhost', 8765) as server: await server.serve_forever() — server; agent creates WebSocket server endpoint
- • Agent broadcast — CONNECTIONS = set(); async def handler(ws): CONNECTIONS.add(ws); try: async for msg in ws: websockets.broadcast(CONNECTIONS, msg); finally: CONNECTIONS.remove(ws); — broadcast; agent forwards messages to all connected clients; broadcast() is non-blocking (fire-and-forget)
- • Agent reconnect — async def connect_with_retry(): while True: try: async with websockets.connect(uri) as ws: async for msg in ws: handle(msg); except (websockets.ConnectionClosed, OSError): await asyncio.sleep(5) — reconnect; agent implements reconnection loop; ConnectionClosed when server closes; OSError for network failure
- • Agent ping/pong keepalive — async with websockets.connect(uri, ping_interval=20, ping_timeout=10) as ws: async for msg in ws: handle(msg) — keepalive; agent configures automatic heartbeats; ping_interval=20 sends ping every 20s; ping_timeout=10 closes if pong not received within 10s; prevents proxy timeout
Not For
- • HTTP REST APIs — websockets is WebSocket-only; for REST use FastAPI/httpx
- • Built-in reconnection — websockets has no automatic reconnect; implement retry loop manually
- • High-throughput message routing — for complex routing use aiohttp WebSocket or FastAPI WebSocket with message routing
Interface
Authentication
No built-in auth. Basic HTTP auth via URI: wss://user:pass@host/ws. Custom auth via headers: additional_headers={'Authorization': 'Bearer token'}. SSL/TLS via ssl= parameter.
Pricing
websockets is BSD 3-Clause licensed. Free for all use.
Agent Metadata
Known Gotchas
- ⚠ recv() raises ConnectionClosed on disconnect — async for msg in ws: iterates until connection closes; manually: msg = await ws.recv() raises ConnectionClosedOK (clean) or ConnectionClosedError (error); agent code: use async for in a try/except ConnectionClosed block; or check ws.closed before recv(); ConnectionClosed.code and .reason for diagnosis
- ⚠ send() does not guarantee delivery — websockets.send() puts message in buffer; network failure after send = message lost; WebSocket protocol has no ack; agent code: implement application-level ack if delivery guarantee needed; ping/pong only tests connection liveness, not message delivery
- ⚠ broadcast() is NOT awaited — websockets.broadcast(connections, message) is synchronous call (not coroutine); does NOT wait for all clients to receive; failed sends raise exceptions that must be caught; agent code: broadcast(connections, msg) — no await; wrap in try/except if any connection failure should be handled
- ⚠ SSL in production — websockets.connect('wss://...') auto-uses default SSL context; custom certs: ssl=ssl.create_default_context(cafile='ca.crt'); server: websockets.serve(handler, ssl=ssl_context); agent code: always use wss:// for production; validate server certificate; ssl=False only for testing
- ⚠ Connection object not thread-safe — websockets Connection must only be used from asyncio event loop it was created in; passing to threads: must use asyncio.run_coroutine_threadsafe(); agent code: all websocket operations must be coroutines; to send from sync thread: loop.call_soon_threadsafe(asyncio.ensure_future, ws.send(msg))
- ⚠ Concurrent recv() not supported — only one coroutine may await ws.recv() at a time; multiple consumers need producer-consumer pattern: async def producer(ws, queue): async for msg in ws: await queue.put(msg); agent code: single consumer per connection; use asyncio.Queue to distribute to multiple processors; do not await ws.recv() from multiple coroutines
Alternatives
Full Evaluation Report
Detailed scoring breakdown, competitive positioning, security analysis, and improvement recommendations for websockets.
Scores are editorial opinions as of 2026-03-06.