websockets is a pure-asyncio WebSocket library for building servers and clients. pip install websockets. Server: from websockets.server import serve; async with serve(handler, "localhost", 8765): await asyncio.Future(). Handler: async def handler(ws): msg = await ws.recv(); await ws.send(f"Echo: {msg}"). Client: from websockets.sync.client import connect; with connect("ws://localhost:8765") as ws: ws.send("Hello"); print(ws.recv()). Async client: from websockets.asyncio.client import connect; async with connect("ws://...") as ws: await ws.send("hi"); msg = await ws.recv(). Broadcast: for ws in connected: await ws.send(message). JSON: await ws.send(json.dumps(data)); data = json.loads(await ws.recv()). ConnectionClosed: from websockets.exceptions import ConnectionClosed. Close: await ws.close(code=1000, reason="done"). Ping: await ws.ping(); pong = await ws.ping(). SSL: ssl.SSLContext passed to serve(ssl=ctx). connect(uri, ssl=ctx). Max size: serve(..., max_size=2**20). Compression: serve(..., compression="deflate"). Origin: serve(..., origins=["https://example.com"]). Headers: ws.request.headers. Path: ws.request.path. ws.remote_address. Subprotocol: serve(..., subprotocols=["chat"]). Timeout: async with asyncio.timeout(5): await ws.recv(). websockets.asyncio.server. Claude Code generates websockets echo servers, real-time broadcast systems, and async WebSocket client libraries.
CLAUDE.md for websockets
## websockets Stack
- Version: websockets >= 13.0 | pip install websockets
- Server: async with serve(handler, host, port): await asyncio.Future()
- Handler: async def handler(ws): msg=await ws.recv(); await ws.send(reply)
- Async client: async with connect("ws://host:port/path") as ws: ...
- Sync client: with websockets.sync.client.connect("ws://...") as ws: ...
- JSON: json.dumps/loads around send/recv
- Error: catch ConnectionClosed for graceful disconnect handlingwebsockets Real-Time Pipeline
# app/ws.py — websockets echo server, broadcast, auth, client, and room manager
from __future__ import annotations
import asyncio
import json
import logging
from typing import Any, Callable, Awaitable
from websockets.asyncio.client import connect
from websockets.asyncio.server import ServerConnection, serve
from websockets.exceptions import ConnectionClosed
log = logging.getLogger(__name__)
# ─────────────────────────────────────────────────────────────────────────────
# 1. Basic echo server
# ─────────────────────────────────────────────────────────────────────────────
async def echo_handler(ws: ServerConnection) -> None:
"""
Simple echo server handler — reflect every message back.
Handles text and binary frames; closes cleanly on disconnect.
"""
remote = ws.remote_address
log.info("Connected: %s", remote)
try:
async for message in ws:
if isinstance(message, str):
await ws.send(f"Echo: {message}")
else:
await ws.send(message) # binary echo
except ConnectionClosed:
log.info("Disconnected: %s", remote)
async def run_echo_server(host: str = "localhost", port: int = 8765) -> None:
"""Run a WebSocket echo server until interrupted."""
async with serve(echo_handler, host, port) as server:
log.info("Echo server on ws://%s:%d", host, port)
await server.serve_forever()
# ─────────────────────────────────────────────────────────────────────────────
# 2. Broadcast (pub-sub / chat room)
# ─────────────────────────────────────────────────────────────────────────────
class ConnectionHub:
"""
A hub that tracks connected WebSocket clients and supports broadcasting.
Usage:
hub = ConnectionHub()
async def handler(ws):
async with hub.connected(ws):
async for msg in ws:
await hub.broadcast(msg, exclude=ws)
"""
def __init__(self):
self._connections: set[ServerConnection] = set()
@property
def count(self) -> int:
return len(self._connections)
async def _add(self, ws: ServerConnection) -> None:
self._connections.add(ws)
async def _remove(self, ws: ServerConnection) -> None:
self._connections.discard(ws)
class _CM:
def __init__(self, hub: "ConnectionHub", ws: ServerConnection):
self._hub = hub
self._ws = ws
async def __aenter__(self):
await self._hub._add(self._ws)
return self._ws
async def __aexit__(self, *_):
await self._hub._remove(self._ws)
def connected(self, ws: ServerConnection):
"""Context manager: register ws on enter, deregister on exit."""
return self._CM(self, ws)
async def broadcast(
self,
message: str | bytes,
exclude: ServerConnection | None = None,
) -> int:
"""
Send message to all connected clients.
exclude: skip this connection (e.g. the sender).
Returns count of successful sends.
"""
targets = self._connections - ({exclude} if exclude else set())
sent = 0
for ws in list(targets):
try:
await ws.send(message)
sent += 1
except ConnectionClosed:
self._connections.discard(ws)
return sent
async def broadcast_json(self, data: Any, exclude: ServerConnection | None = None) -> int:
return await self.broadcast(json.dumps(data), exclude=exclude)
# ─────────────────────────────────────────────────────────────────────────────
# 3. JSON message protocol
# ─────────────────────────────────────────────────────────────────────────────
async def send_json(ws: ServerConnection, data: Any) -> None:
"""Send a Python object as JSON."""
await ws.send(json.dumps(data))
async def recv_json(ws: ServerConnection) -> Any:
"""Receive the next message and parse as JSON."""
raw = await ws.recv()
return json.loads(raw)
class JsonProtocol:
"""
Wrapper for a WebSocket connection with typed message dispatch.
Usage:
proto = JsonProtocol(ws)
await proto.send({"type": "hello", "name": "Alice"})
msg = await proto.recv()
if msg["type"] == "ping":
await proto.send({"type": "pong"})
"""
def __init__(self, ws: ServerConnection):
self._ws = ws
async def send(self, data: Any) -> None:
await self._ws.send(json.dumps(data))
async def recv(self) -> Any:
raw = await self._ws.recv()
return json.loads(raw)
async def handle_loop(
self,
dispatch: Callable[[Any], Awaitable[Any | None]],
) -> None:
"""
Read messages in a loop, passing each to dispatch(msg).
If dispatch returns a non-None value, send it back.
"""
try:
async for raw in self._ws:
msg = json.loads(raw)
reply = await dispatch(msg)
if reply is not None:
await self.send(reply)
except ConnectionClosed:
pass
# ─────────────────────────────────────────────────────────────────────────────
# 4. Room-based chat server
# ─────────────────────────────────────────────────────────────────────────────
class ChatRoom:
"""
Named chat room: join/leave/broadcast with message history.
Usage:
room_manager = {room_name: ChatRoom(name)}
async def ws_handler(ws):
room = rooms["general"]
await room.join(ws, username="Alice")
"""
def __init__(self, name: str, max_history: int = 50):
self.name = name
self._members: dict[ServerConnection, str] = {} # ws → username
self._history: list[dict] = []
self._max_history = max_history
@property
def member_count(self) -> int:
return len(self._members)
async def join(self, ws: ServerConnection, username: str) -> None:
"""Add ws to the room and announce."""
self._members[ws] = username
join_msg = {"type": "join", "user": username, "room": self.name}
await self._broadcast(json.dumps(join_msg), exclude=ws)
# Send history to the new member
for msg in self._history:
try:
await ws.send(json.dumps(msg))
except ConnectionClosed:
break
async def leave(self, ws: ServerConnection) -> None:
"""Remove ws from the room."""
username = self._members.pop(ws, "unknown")
leave_msg = {"type": "leave", "user": username, "room": self.name}
await self._broadcast(json.dumps(leave_msg))
async def send_message(self, ws: ServerConnection, text: str) -> None:
"""Broadcast a chat message from ws."""
username = self._members.get(ws, "anon")
msg = {"type": "message", "user": username, "room": self.name, "text": text}
self._history.append(msg)
if len(self._history) > self._max_history:
self._history.pop(0)
await self._broadcast(json.dumps(msg))
async def _broadcast(self, raw: str, exclude: ServerConnection | None = None) -> None:
for ws in list(self._members):
if ws is not exclude:
try:
await ws.send(raw)
except ConnectionClosed:
self._members.pop(ws, None)
# ─────────────────────────────────────────────────────────────────────────────
# 5. Async client helpers
# ─────────────────────────────────────────────────────────────────────────────
async def ws_send_receive(uri: str, messages: list[Any], timeout: float = 5.0) -> list[Any]:
"""
Connect to uri, send each message (as JSON), collect replies.
Returns list of parsed JSON responses.
"""
responses = []
async with connect(uri) as ws:
for msg in messages:
await ws.send(json.dumps(msg))
try:
async with asyncio.timeout(timeout):
raw = await ws.recv()
responses.append(json.loads(raw))
except TimeoutError:
break
return responses
async def ws_subscribe(
uri: str,
on_message: Callable[[Any], Awaitable[None]],
reconnect: bool = True,
reconnect_delay: float = 2.0,
) -> None:
"""
Subscribe to a WebSocket stream; call on_message for each JSON message.
reconnect=True: automatically reconnect on disconnect.
"""
while True:
try:
async with connect(uri) as ws:
log.info("Subscribed to %s", uri)
async for raw in ws:
await on_message(json.loads(raw))
except ConnectionClosed as e:
log.warning("Disconnected from %s: %s", uri, e)
except Exception as e:
log.error("Error from %s: %s", uri, e)
if not reconnect:
break
log.info("Reconnecting in %.1fs...", reconnect_delay)
await asyncio.sleep(reconnect_delay)
# ─────────────────────────────────────────────────────────────────────────────
# Demo
# ─────────────────────────────────────────────────────────────────────────────
async def demo():
"""Run an in-process echo server + client to verify the setup."""
import threading
port = 18765
async def _server():
hub = ConnectionHub()
async def handler(ws):
async with hub.connected(ws):
async for msg in ws:
await hub.broadcast(msg)
async with serve(handler, "localhost", port):
await asyncio.sleep(2)
# Start server in background task
server_task = asyncio.create_task(_server())
await asyncio.sleep(0.1) # let server start
print("=== Client send/receive ===")
replies = await ws_send_receive(
f"ws://localhost:{port}",
[{"text": "Hello"}, {"text": "World"}],
)
print("Replies:", replies)
server_task.cancel()
try:
await server_task
except asyncio.CancelledError:
pass
print("Demo complete")
if __name__ == "__main__":
logging.basicConfig(level=logging.INFO)
asyncio.run(demo())For the socket.io (python-socketio) alternative — socket.io adds namespaces, rooms, acknowledgements, and automatic reconnect at the protocol level; websockets is lower-level and implements the IETF RFC 6455 WebSocket standard directly — use websockets when you want standards compliance and control over your protocol, socket.io when you need its higher-level abstractions or browser client library. For the FastAPI WebSocket alternative — FastAPI’s built-in WebSocket support (@app.websocket("/ws")) wraps Starlette’s WebSocket and integrates with FastAPI’s dependency injection; the standalone websockets library is better for pure async WebSocket microservices or CLI clients where you don’t need an HTTP framework alongside. The Claude Skills 360 bundle includes websockets skill sets covering echo_handler/run_echo_server(), ConnectionHub broadcast manager, JsonProtocol send/recv/handle_loop(), ChatRoom join/leave/send_message with history, send_json/recv_json helpers, ws_send_receive() client, ws_subscribe() streaming client with auto-reconnect, and in-process demo. Start with the free tier to try WebSocket real-time code generation.