Getting Started with Sanic WebSockets

Sanic has gained momentum as a high-performance Python web framework designed with speed and developer experience in mind.

The framework offers a clean, intuitive interface for implementing real-time communication, enabling you to craft responsive applications with features like instant messaging, live notifications, and dynamic content updates.

This practical guide will take you through building WebSocket functionality in Sanic step by step, using patterns that remain effective as your application grows.

Prerequisites

Before exploring Sanic WebSocket development, confirm you have Python 3.8 or newer on your system. Familiarity with Python's async/await syntax will prove beneficial since Sanic embraces asynchronous programming throughout its design.

Understanding JavaScript fundamentals will also help when we create interactive web interfaces to demonstrate our WebSocket functionality.

Establishing your Sanic WebSocket environment

Creating reliable real-time applications begins with effective project organization that clearly separates different concerns. We'll establish a solid foundation to keep WebSocket management well-organized, leveraging Sanic's simple async architecture.

Begin by establishing your development environment with dependency isolation:

mkdir sanic_websockets && cd sanic_websockets

python3 -m venv venv

source venv/bin/activate

These steps create your project workspace, establish an isolated Python environment, and activate it to maintain clean dependency management separate from other projects.

Install Sanic with its comprehensive feature set:

pip install sanic

Sanic distinguishes itself by including WebSocket support directly in the core framework. This integrated approach eliminates the need for additional packages while providing robust real-time communication capabilities without extra configuration overhead.

Establish your application structure with the foundational file:

from sanic import Sanic, Request, Websocket
from sanic.response import html

app = Sanic("realtime_app")

# WebSocket handlers will be implemented here

if __name__ == "__main__":
    app.run(host="0.0.0.0", port=8000, debug=True)

This initial setup provides the core structure for WebSocket development. Sanic's unified handling of both HTTP and WebSocket protocols simplifies development, while debug mode offers valuable development features, including automatic code reloading and comprehensive error reporting.

Sanic treats WebSocket connections with the same importance as HTTP routes, providing a consistent development experience. WebSocket handlers accept two arguments: the HTTP request context and the WebSocket connection instance, giving you access to both request information and bidirectional communication capabilities.

Create a foundational echo server that illustrates essential WebSocket patterns:

from sanic import Sanic, Request, Websocket
from sanic.response import html

app = Sanic("realtime_app")

async def websocket_handler(request: Request, ws: Websocket):
    """Streamlined WebSocket handler using iteration"""
    print(f"Client connected from {request.ip}")

    try:
        async for message in ws:
            print(f"Processing message: {message}")
            await ws.send(f"Server response: {message}")
    except Exception as e:
        print(f"Connection terminated: {e}")

if __name__ == "__main__":
    app.run(host="0.0.0.0", port=8000, debug=True)

The async for construct automatically manages the message reception cycle, creating more maintainable and readable code. This pattern eliminates manual while True loops while delivering identical message-processing functionality.

The iteration concludes automatically when connections close, streamlining connection lifecycle management significantly.

Developing a comprehensive client interface with templates

Professional real-time applications benefit from proper separation of concerns, keeping HTML templates, CSS styles, and JavaScript logic in dedicated files.

Let's restructure your WebSocket interface using Sanic's templating capabilities with organized static assets.

First, install the required dependencies:

pip install jinja2 sanic-jinja2

Create the proper directory structure for your templates and static assets:

mkdir -p templates static/css static/js

Create your main template file with dynamic content placeholders and clean HTML structure:

<!DOCTYPE html>
<html lang="en">
<head>
    <meta charset="UTF-8">
    <meta name="viewport" content="width=device-width, initial-scale=1.0">
    <title>{{ title }}</title>
    <link rel="stylesheet" href="/static/css/style.css">
</head>
<body>
    <div class="container">
        <div class="header">
            <h1>{{ title }}</h1>
            <div>{{ subtitle }}</div>
        </div>

        <div id="statusIndicator" class="status-bar connecting">{{ connection_message }}</div>

        <div id="messageArea"></div>

        <div class="input-section">
            <input 
                type="text" 
                id="messageField" 
                placeholder="{{ input_placeholder }}" 
                maxlength="500"
                disabled
            >
            <button id="submitButton" disabled>{{ button_text }}</button>
        </div>
    </div>

    <script src="/static/js/websocket-client.js"></script>
</body>
</html>

This HTML template uses Jinja2 variables like {{ title }} and {{ subtitle }} for dynamic content that can be customized from your Python code. The template links to external CSS and JavaScript files using /static/ paths, which Sanic will serve automatically. The structure includes a statusIndicator for connection status, messageArea for displaying chat messages, and an input section with a text field and send button. Notice how the input field starts disabled - this prevents users from sending messages before the WebSocket connection is established.

Create your stylesheet with modern design, animations, and responsive layout:

* { 
    margin: 0; 
    padding: 0; 
    box-sizing: border-box; 
}

body { 
    font-family: -apple-system, BlinkMacSystemFont, 'Segoe UI', system-ui, sans-serif;
    background: linear-gradient(135deg, #1e3c72 0%, #2a5298 100%);
    min-height: 100vh;
    display: flex;
    align-items: center;
    justify-content: center;
    padding: 20px;
}

.container {
    background: white;
    border-radius: 16px;
    box-shadow: 0 25px 50px rgba(0,0,0,0.2);
    overflow: hidden;
    width: 100%;
    max-width: 480px;
}

.header {
    background: linear-gradient(135deg, #667eea, #764ba2);
    color: white;
    padding: 24px;
    text-align: center;
}

.header h1 {
    font-size: 1.75rem;
    font-weight: 700;
    margin-bottom: 8px;
}

.status-bar {
    padding: 16px 24px;
    font-size: 14px;
    font-weight: 600;
    text-align: center;
    transition: all 0.3s ease;
}

.status-bar.connecting {
    background: #fef3c7;
    color: #92400e;
}

.status-bar.connected {
    background: #d1fae5;
    color: #065f46;
}

.status-bar.disconnected {
    background: #fee2e2;
    color: #991b1b;
}

#messageArea {
    height: 400px;
    overflow-y: auto;
    padding: 24px;
    display: flex;
    flex-direction: column;
    gap: 16px;
    background: #f8fafc;
}

.message {
    max-width: 85%;
    padding: 14px 18px;
    border-radius: 20px;
    word-wrap: break-word;
    font-size: 15px;
    line-height: 1.5;
    animation: messageSlide 0.3s ease-out;
}

@keyframes messageSlide {
    from {
        opacity: 0;
        transform: translateY(10px);
    }
    to {
        opacity: 1;
        transform: translateY(0);
    }
}

.message.outbound {
    background: linear-gradient(135deg, #667eea, #764ba2);
    color: white;
    align-self: flex-end;
    margin-left: auto;
}

.message.inbound {
    background: white;
    color: #1f2937;
    align-self: flex-start;
    border: 1px solid #e5e7eb;
    box-shadow: 0 2px 4px rgba(0,0,0,0.05);
}

.message.notification {
    background: #eff6ff;
    color: #1e40af;
    align-self: center;
    text-align: center;
    font-style: italic;
    font-size: 13px;
    border: 1px solid #dbeafe;
}

.input-section {
    padding: 24px;
    border-top: 1px solid #f3f4f6;
    display: flex;
    gap: 16px;
    background: white;
}

#messageField {
    flex: 1;
    padding: 14px 18px;
    border: 2px solid #e5e7eb;
    border-radius: 28px;
    font-size: 16px;
    outline: none;
    transition: border-color 0.3s ease;
}

#messageField:focus {
    border-color: #667eea;
    box-shadow: 0 0 0 3px rgba(102, 126, 234, 0.1);
}

#submitButton {
    padding: 14px 28px;
    background: linear-gradient(135deg, #667eea, #764ba2);
    color: white;
    border: none;
    border-radius: 28px;
    font-size: 16px;
    font-weight: 600;
    cursor: pointer;
    transition: transform 0.2s ease, opacity 0.2s ease;
}

#submitButton:hover:not(:disabled) {
    transform: translateY(-1px);
    box-shadow: 0 4px 12px rgba(102, 126, 234, 0.3);
}

#submitButton:disabled {
    opacity: 0.4;
    cursor: not-allowed;
    transform: none;
}

/* Responsive design */
@media (max-width: 640px) {
    body {
        padding: 10px;
    }

    .container {
        max-width: 100%;
    }

    .header {
        padding: 20px;
    }

    .header h1 {
        font-size: 1.5rem;
    }

    #messageArea {
        height: 300px;
        padding: 16px;
    }

    .input-section {
        padding: 16px;
    }
}

This stylesheet creates a modern chat interface with professional styling and user experience enhancements.

The .status-bar classes use different background colors to indicate connection states - yellow for connecting, green for connected, and red for disconnected.

The @keyframes messageSlide animation makes new messages appear smoothly, sliding up from below with a fade-in effect. Notice how .message.outbound uses align-self: flex-end to position sent messages on the right, while .message.inbound aligns received messages to the left.

Create your JavaScript WebSocket client with advanced features like reconnection and error handling:

class SanicWebSocketClient {
    constructor() {
        this.socket = null;
        this.reconnectAttempts = 0;
        this.maxReconnectAttempts = 3;
        this.reconnectDelay = 1000;

        // Get DOM elements
        this.messageField = document.getElementById('messageField');
        this.submitButton = document.getElementById('submitButton');
        this.messageArea = document.getElementById('messageArea');
        this.statusIndicator = document.getElementById('statusIndicator');

        this.initializeEventHandlers();
        this.establishConnection();
    }

    initializeEventHandlers() {
        // Send button click handler
        this.submitButton.addEventListener('click', () => this.transmitMessage());

        // Enter key handler for message input
        this.messageField.addEventListener('keypress', (event) => {
            if (event.key === 'Enter' && !this.submitButton.disabled) {
                this.transmitMessage();
            }
        });

        // Enable/disable send button based on input
        this.messageField.addEventListener('input', (event) => {
            this.submitButton.disabled = !event.target.value.trim() || 
                this.socket?.readyState !== WebSocket.OPEN;
        });

        // Handle page visibility changes
        document.addEventListener('visibilitychange', () => {
            if (document.visibilityState === 'visible' && 
                (!this.socket || this.socket.readyState === WebSocket.CLOSED)) {
                this.establishConnection();
            }
        });
    }

    establishConnection() {
        const protocol = window.location.protocol === 'https:' ? 'wss:' : 'ws:';
        const socketUrl = `${protocol}//${window.location.host}/ws`;

        try {
            this.socket = new WebSocket(socketUrl);
            this.setupWebSocketHandlers();
        } catch (error) {
            console.error('Failed to create WebSocket connection:', error);
            this.handleConnectionError();
        }
    }

    setupWebSocketHandlers() {
        this.socket.onopen = () => {
            this.reconnectAttempts = 0;
            this.updateConnectionStatus('connected', 'Connected to Sanic server');
            this.messageField.disabled = false;
            this.messageField.focus();
            this.displayMessage('Connected to Sanic WebSocket server!', 'notification');
        };

        this.socket.onmessage = (event) => {
            try {
                // Try parsing as JSON first
                const data = JSON.parse(event.data);
                this.handleStructuredMessage(data);
            } catch {
                // Fall back to plain text
                this.displayMessage(event.data, 'inbound');
            }
        };

        this.socket.onclose = (event) => {
            this.updateConnectionStatus('disconnected', 'Connection closed');
            this.messageField.disabled = true;
            this.submitButton.disabled = true;

            if (event.code !== 1000 && this.reconnectAttempts < this.maxReconnectAttempts) {
                this.attemptReconnection();
            } else if (event.code !== 1000) {
                this.displayMessage('Connection lost. Please refresh to reconnect.', 'notification');
            }
        };

        this.socket.onerror = (error) => {
            console.error('WebSocket error:', error);
            this.handleConnectionError();
        };
    }

    handleStructuredMessage(data) {
        switch (data.type) {
            case 'welcome':
                this.displayMessage(data.content, 'notification');
                break;
            case 'broadcast':
                this.displayMessage(data.content, 'inbound');
                this.updateClientCount(data.active_clients);
                break;
            default:
                this.displayMessage(data.content || data.message, 'inbound');
        }
    }

    attemptReconnection() {
        this.reconnectAttempts++;
        this.updateConnectionStatus('connecting', `Reconnecting... (${this.reconnectAttempts}/${this.maxReconnectAttempts})`);

        setTimeout(() => {
            this.establishConnection();
        }, this.reconnectDelay * this.reconnectAttempts);
    }

    handleConnectionError() {
        this.updateConnectionStatus('disconnected', 'Connection failed');
        this.displayMessage('Unable to connect to server', 'notification');
    }

    updateConnectionStatus(status, message) {
        this.statusIndicator.className = `status-bar ${status}`;
        this.statusIndicator.textContent = message;
    }

    updateClientCount(count) {
        if (count > 1) {
            this.updateConnectionStatus('connected', `Connected (${count} clients online)`);
        }
    }

    displayMessage(content, messageType) {
        const messageElement = document.createElement('div');
        messageElement.className = `message ${messageType}`;
        messageElement.textContent = content;

        this.messageArea.appendChild(messageElement);
        this.messageArea.scrollTop = this.messageArea.scrollHeight;

        // Add subtle animation
        messageElement.style.transform = 'translateY(10px)';
        messageElement.style.opacity = '0';

        requestAnimationFrame(() => {
            messageElement.style.transform = 'translateY(0)';
            messageElement.style.opacity = '1';
        });
    }

    transmitMessage() {
        const messageContent = this.messageField.value.trim();

        if (!messageContent || this.socket?.readyState !== WebSocket.OPEN) {
            return;
        }

        this.displayMessage(messageContent, 'outbound');
        this.socket.send(messageContent);
        this.messageField.value = '';
        this.submitButton.disabled = true;
    }

    // Public method to close connection cleanly
    disconnect() {
        if (this.socket && this.socket.readyState === WebSocket.OPEN) {
            this.socket.close(1000);
        }
    }
}

// Initialize WebSocket client when DOM is ready
document.addEventListener('DOMContentLoaded', () => {
    window.webSocketClient = new SanicWebSocketClient();
});

// Clean up on page unload
window.addEventListener('beforeunload', () => {
    if (window.webSocketClient) {
        window.webSocketClient.disconnect();
    }
});

This JavaScript class provides a production-ready WebSocket client with sophisticated features. The constructor initializes reconnection parameters like maxReconnectAttempts = 3 and reconnectDelay = 1000, ensuring the client automatically tries to reconnect if the connection drops.

The establishConnection() method intelligently detects whether to use ws:// or wss:// protocols based on the current page protocol. In setupWebSocketHandlers(), the onmessage event handler uses a try-catch block to first attempt parsing messages as JSON, then falls back to plain text - this allows the same client to handle both simple echo messages and structured data from broadcasting features.

The attemptReconnection() method implements exponential backoff by multiplying the delay by the attempt number, preventing the client from overwhelming a recovering server. The transmitMessage() method includes validation to ensure messages aren't sent when the connection isn't ready, using this.socket?.readyState !== WebSocket.OPEN for safe checking.

Now update your main application file to use templates and serve static files:

from sanic import Sanic, Request, Websocket
from sanic_jinja2 import SanicJinja2
from jinja2 import FileSystemLoader
import asyncio
import logging

# Configure logging
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)

app = Sanic("realtime_app")

# Setup Jinja2 templating
jinja = SanicJinja2(app, loader=FileSystemLoader('templates'))

# Serve static files (CSS, JS, images)
app.static('/static', './static')

@app.route("/")
async def index(request: Request):
    """Render the WebSocket client interface using templates"""
    return jinja.render('index.html', request,
        title="Sanic WebSocket",
        subtitle="Real-Time Communication",
        connection_message="Establishing connection...",
        input_placeholder="Enter your message...",
        button_text="Send"
    )

@app.websocket("/ws")
async def websocket_handler(request: Request, ws: Websocket):
    """Streamlined WebSocket handler using iteration"""
    client_address = request.ip
    logger.info(f"WebSocket connection established from {client_address}")

    try:
        # Send welcome message
        await ws.send("Welcome to Sanic WebSocket server!")

        # Process incoming messages
        async for message in ws:
            logger.info(f"Received from {client_address}: {message}")

            # Generate and send response
            response_message = f"Server response: {message}"
            await ws.send(response_message)
    except asyncio.CancelledError:
        logger.info(f"WebSocket connection cancelled for {client_address}")
    except Exception as error:
        logger.error(f"WebSocket error for {client_address}: {error}")
    finally:
        logger.info(f"WebSocket connection closed for {client_address}")

if __name__ == "__main__":
    app.run(
        host="0.0.0.0", 
        port=8000, 
        debug=True,
        access_log=True
    )

This enhanced application integrates several important components for professional WebSocket development.

The SanicJinja2 import and setup enables template rendering, while FileSystemLoader('templates') tells Jinja2 where to find template files.

The app.static('/static', './static') line creates a route that serves all files from the static directory - when browsers request /static/css/style.css, Sanic automatically serves the file from ./static/css/style.css.

The jinja.render() method in the index route passes template variables as keyword arguments, making the interface customizable without hardcoding text.

The WebSocket handler now includes comprehensive logging with logger.info() calls to track connection lifecycle and message flow. The specific handling of asyncio.CancelledError is crucial - this exception occurs when clients disconnect unexpectedly (like closing a browser tab), and treating it separately from other exceptions prevents false error reports in your logs.

Testing your WebSocket implementation

Launch your Sanic application and verify the real-time communication:

python app.py

The server will display startup information confirming successful initialization:

Main  11:05:29 DEBUG:  Creating multiprocessing context using 'spawn'
2025-08-20 11:05:29,077 - DEBUG - Creating multiprocessing context using 'spawn'
Main  11:05:29 DEBUG:  Starting a process: Sanic-Server-0-0
2025-08-20 11:05:29,078 - DEBUG - Starting a process: Sanic-Server-0-0
Srv 0 11:05:29 DEBUG:  Process ack: Sanic-Server-0-0 [87154]
2025-08-20 11:05:29,210 - DEBUG - Process ack: Sanic-Server-0-0 [87154]
Srv 0 11:05:29 INFO:   Starting worker [87154]
2025-08-20 11:05:29,212 - INFO - Starting worker [87154]

Open your browser to http://localhost:8000. The connection status will transition from "Establishing connection..." to "Connected to Sanic server" as the WebSocket connection initializes successfully:

The interface displays a welcome message immediately upon connection, confirming your WebSocket handler functions correctly. Notice how the status bar changes from yellow "Establishing connection..." to green "Connected to Sanic server", indicating a successful WebSocket handshake. The input field becomes enabled and focuses automatically, ready for user interaction.

Enter messages in the input field and press Enter to observe instant echo responses, demonstrating bidirectional communication capabilities:

Type any message like "test message" and press Enter. The message appears immediately on the right side as an outbound message (styled with the gradient background), while the server's echo response appears on the left as an inbound message:

This demonstrates the complete WebSocket communication cycle - your message travels from the client to the server through the WebSocket connection, gets processed by the websocket_handler() function, and returns as an echo response.

Final thoughts

This guide took you through building WebSocket functionality in Sanic, starting with basic echo servers and advancing to professional applications with modern templates, styling, and error handling.

Sanic's built-in WebSocket support makes real-time development straightforward by eliminating the complexity found in other frameworks. You can focus on your application logic while Sanic handles the infrastructure. The async/await patterns work naturally with modern Python development, and the framework's performance makes it excellent for high-traffic real-time applications.

As your applications grow, you can extend this foundation with user authentication, message persistence using databases like PostgreSQL or Redis, and horizontal scaling with message brokers. The Sanic documentation provides detailed guidance on security, performance optimization, and deployment best practices.

Got an article suggestion? Let us know

Next article

Get Started with Job Scheduling in Python

Learn how to create and monitor Python scheduled tasks in a production environment

→

This work is licensed under a Creative Commons Attribution-NonCommercial-ShareAlike 4.0 International License.