WebSocket Skill

Production-grade WebSocket patterns for real-time communication in Python applications.

When to Use This Skill

Use this skill when:

• •Building real-time features (chat, notifications, live updates)
• •Implementing bidirectional client-server communication
• •Creating collaborative features (multi-user editing, presence)
• •Building dashboards with live data feeds
• •Implementing game servers or real-time applications

Core Mental Model

A WebSocket connection is fundamentally different from HTTP:

HTTP:     Request → Response → Done
WebSocket: Connect → Accept → [Receive Loop] → Disconnect → Cleanup

The receive loop is unavoidable because:

1. •The connection is long-lived - something must continuously listen
2. •Messages arrive asynchronously from the client
3. •The loop defines the connection's lifetime boundary
4. •Even when frameworks abstract it, a loop exists underneath

Architecture Overview

┌─────────────────────────────────────────────────────────────┐
│                     WebSocket System                         │
├─────────────────────────────────────────────────────────────┤
│                                                             │
│  ┌──────────────┐    ┌──────────────────────────────────┐  │
│  │   Client 1   │───▶│                                  │  │
│  └──────────────┘    │                                  │  │
│                      │     Connection Manager           │  │
│  ┌──────────────┐    │  ┌─────────────────────────────┐ │  │
│  │   Client 2   │───▶│  │ connections: Set[WebSocket] │ │  │
│  └──────────────┘    │  │ rooms: Dict[str, Set[WS]]   │ │  │
│                      │  └─────────────────────────────┘ │  │
│  ┌──────────────┐    │                                  │  │
│  │   Client N   │───▶│  connect() → track              │  │
│  └──────────────┘    │  disconnect() → remove          │  │
│                      │  broadcast() → fan-out          │  │
│                      │  send_to_room() → targeted      │  │
│                      └──────────────────────────────────┘  │
│                                                             │
└─────────────────────────────────────────────────────────────┘

File Structure

app/
├── websocket/
│   ├── __init__.py
│   ├── manager.py          # ConnectionManager class
│   ├── handlers.py         # Message handlers by type
│   ├── models.py           # Message schemas (Pydantic)
│   └── router.py           # WebSocket endpoint
└── main.py                 # Mount WebSocket router

Core Principles

1. Explicit Connection Lifecycle

Connections must be:

• •Accepted explicitly - Never assume connection success
• •Tracked centrally - Single source of truth
• •Removed on disconnect - Eager cleanup prevents leaks
• •Silent failures are worse than explicit disconnects

2. Centralized Connection Management

# Connection state lives in ONE place
class ConnectionManager:
    def __init__(self):
        self.active_connections: set[WebSocket] = set()

    async def connect(self, websocket: WebSocket) -> None:
        await websocket.accept()
        self.active_connections.add(websocket)

    def disconnect(self, websocket: WebSocket) -> None:
        self.active_connections.discard(websocket)

Business logic must NEVER handle socket tracking directly.

3. Receive Loop Responsibility

The receive loop:

• •Defines connection lifetime
• •Is the only entry point for client messages
• •Must handle disconnection gracefully

async def websocket_endpoint(websocket: WebSocket):
    await manager.connect(websocket)
    try:
        while True:  # THE LOOP
            data = await websocket.receive_text()
            # validate → dispatch → respond
    except WebSocketDisconnect:
        pass  # Expected, not exceptional
    finally:
        manager.disconnect(websocket)  # Always cleanup

4. Message Contract Discipline

Every message must be:

• •Structured: JSON, not raw strings
• •Typed: {"type": "chat", "payload": {...}}
• •Intent-declaring: The type field routes to handlers

class WebSocketMessage(BaseModel):
    type: str
    payload: dict = {}

5. Defensive Failure Handling

• •Disconnections happen constantly (normal, not exceptional)
• •Writes can fail even after reads succeed
• •Dead connections must be cleaned eagerly
• •One failure must not cascade to others

Extended Concepts

Rooms / Channels

A room is a logical routing layer, not a socket feature:

class ConnectionManager:
    def __init__(self):
        self.connections: set[WebSocket] = set()
        self.rooms: dict[str, set[WebSocket]] = defaultdict(set)

    def join_room(self, websocket: WebSocket, room_id: str) -> None:
        self.rooms[room_id].add(websocket)

    def leave_room(self, websocket: WebSocket, room_id: str) -> None:
        self.rooms[room_id].discard(websocket)

    async def broadcast_to_room(self, room_id: str, message: str) -> None:
        for connection in self.rooms[room_id]:
            await connection.send_text(message)

Key insights:

• •Connection belongs to exactly 1 manager, but 0+ rooms
• •Room membership is explicit (join/leave operations)
• •Leaving a room ≠ disconnecting (orthogonal concerns)

Performance Mental Model

Performance is about control, not speed:

Safety principles:

1. •Never block the receive loop with slow operations
2. •Writes must not delay reads
3. •One slow client must not affect others

Generation Order

When implementing WebSocket features:

1. •Define message types - What messages will be exchanged?
2. •Create ConnectionManager - Central state management
3. •Implement receive loop - WebSocket endpoint
4. •Add message handlers - Route by message type
5. •Add rooms (if needed) - Logical grouping
6. •Add error handling - Defensive cleanup

Quick Reference

# Accept connection
await websocket.accept()

# Receive message
data = await websocket.receive_text()
data = await websocket.receive_json()

# Send message
await websocket.send_text(message)
await websocket.send_json(data)

# Close connection
await websocket.close(code=1000)

# Handle disconnection
from starlette.websockets import WebSocketDisconnect

What "Production-Grade" Means

Not:

• •Feature-rich
• •Horizontally scalable (yet)
• •Prematurely optimized

Yes:

• •Predictable behavior under all conditions
• •Clear separation: connection management | business logic | routing
• •Safe defaults (cleanup on failure, structured messages)
• •Clear upgrade paths to rooms, scaling, tuning

Explicit Non-Goals

These are intentionally excluded to keep the core pattern clean:

• •Authentication / authorization (handle before WebSocket upgrade)
• •Persistence (message history is a separate concern)
• •Distributed state (requires Redis/pubsub, out of scope)
• •Message ordering guarantees across processes

References

See the references/ directory for:

• •connection-manager-pattern.md - Centralized connection tracking
• •receive-loop-pattern.md - Message loop structure and lifetime
• •message-protocol-pattern.md - JSON message contracts
• •rooms-pattern.md - Logical routing and channels
• •error-handling-pattern.md - Defensive failure handling