Skip to content

WebSocket Clients: Python

This guide covers how to create and manage WebSocket clients in Python, suitable for backend services, scripts, or CLI tools that interact with WebSocket servers.

🚀 1. Using websockets Library

The websockets library is a popular choice for Python WebSocket clients and servers.

Installation

pip install websockets

🏗 2. Basic Client Example

import asyncio
import websockets
import json

async def websocket_client():
    uri = "wss://your-api.com/ws?token=YOUR_TOKEN"
    async with websockets.connect(uri) as websocket:
        # Send a message
        message = {
            "type": "auth",
            "data": {"token": "YOUR_TOKEN"}
        }
        await websocket.send(json.dumps(message))

        # Receive messages
        async for message in websocket:
            msg = json.loads(message)
            await handle_message(msg)

async def handle_message(msg):
    event_type = msg.get("type")
    data = msg.get("data", {})
    if event_type == "connection_established":
        print(f"Connected as user {data.get('user_id')}")
    elif event_type == "new_message":
        print(f"New message: {data.get('message')}")
    elif event_type == "error":
        print(f"Error {data.get('code')}: {data.get('message')}")
    else:
        print(f"Unhandled message: {msg}")

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

🔁 3. Reconnection Logic with Exponential Backoff

import asyncio
import json
import websockets
import random

async def connect_with_retries(uri, max_retries=5):
    retries = 0
    while retries < max_retries:
        try:
            async with websockets.connect(uri) as websocket:
                print("WebSocket connected")
                await websocket.send(json.dumps({"type": "auth", "data": {"token": "YOUR_TOKEN"}}))
                async for message in websocket:
                    msg = json.loads(message)
                    await handle_message(msg)
        except (websockets.ConnectionClosed, ConnectionRefusedError) as e:
            wait_time = (2 ** retries) + random.random()
            print(f"Connection lost: {e}. Reconnecting in {wait_time:.1f} seconds...")
            await asyncio.sleep(wait_time)
            retries += 1
    print("Max retries reached. Exiting.")

async def handle_message(msg):
    print(f"Received message: {msg}")

if __name__ == "__main__":
    uri = "wss://your-api.com/ws?token=YOUR_TOKEN"
    asyncio.run(connect_with_retries(uri))

🔐 4. Authentication

Pass tokens either via query parameters in the URI or send an authentication message immediately after connecting.

# Option 1: Token in URI (shown above)

# Option 2: Send auth message after connecting
await websocket.send(json.dumps({
    "type": "auth",
    "data": {"token": "YOUR_TOKEN"}
}))

🧠 5. Handling Server Messages

Implement a dispatcher based on the type field:

async def handle_message(msg):
    event_type = msg.get("type")
    data = msg.get("data", {})

    if event_type == "connection_established":
        print(f"Connected as user {data.get('user_id')}")
    elif event_type == "new_message":
        print(f"Message: {data.get('message')}")
    elif event_type == "error":
        print(f"Error {data.get('code')}: {data.get('message')}")
    else:
        print(f"Unknown event type: {event_type}")

🧪 6. Debugging Tips

  • Enable logging for the websockets library:
import logging
logging.basicConfig(level=logging.DEBUG)
  • Use Wireshark or other network tools to inspect traffic.
  • Handle exceptions gracefully to avoid client crashes.
  • Test with invalid messages to ensure error handling works.

✅ 7. Best Practices

  • Use secure WebSocket connections (wss://) in production.
  • Implement reconnect logic with backoff to handle network issues.
  • Validate and sanitize incoming messages.
  • Log errors and connection events for observability.
  • Cleanly close connections when shutting down.

🔚 Next Steps