Skip to content

Instantly share code, notes, and snippets.

What would you like to do?
Simple websocket server in Python. Echos back whatever is received. Works with Chome, Firefox 16, IE 10.
import time
import struct
import socket
import hashlib
import base64
import sys
from select import select
import re
import logging
from threading import Thread
import signal
# Simple WebSocket server implementation. Handshakes with the client then echos back everything
# that is received. Has no dependencies (doesn't require Twisted etc) and works with the RFC6455
# version of WebSockets. Tested with FireFox 16, though should work with the latest versions of
# IE, Chrome etc.
# Adapted from with various functions stolen from other sites, see
# below for full details.
# Constants
MAGIC-GUID = "258EAFA5-E914-47DA-95CA-C5AB0DC85B11"
TEXT = 0x01
BINARY = 0x02
# WebSocket implementation
class WebSocket(object):
handshake = (
"HTTP/1.1 101 Web Socket Protocol Handshake\r\n"
"Upgrade: WebSocket\r\n"
"Connection: Upgrade\r\n"
"Sec-WebSocket-Accept: %(acceptstring)s\r\n"
"Server: TestTest\r\n"
"Access-Control-Allow-Origin: http://localhost\r\n"
"Access-Control-Allow-Credentials: true\r\n"
# Constructor
def __init__(self, client, server):
self.client = client
self.server = server
self.handshaken = False
self.header = "" = ""
# Serve this client
def feed(self, data):
# If we haven't handshaken yet
if not self.handshaken:
logging.debug("No handshake yet")
self.header += data
if self.header.find('\r\n\r\n') != -1:
parts = self.header.split('\r\n\r\n', 1)
self.header = parts[0]
if self.dohandshake(self.header, parts[1]):"Handshake successful")
self.handshaken = True
# We have handshaken
logging.debug("Handshake is complete")
# Decode the data that we received according to section 5 of RFC6455
recv = self.decodeCharArray(data)
# Send our reply
# Stolen from
def sendMessage(self, s):
Encode and send a WebSocket message
# Empty message to start with
message = ""
# always send an entire message as one frame (fin)
b1 = 0x80
# in Python 2, strs are bytes and unicodes are strings
if type(s) == unicode:
b1 |= TEXT
payload = s.encode("UTF8")
elif type(s) == str:
b1 |= TEXT
payload = s
# Append 'FIN' flag to the message
message += chr(b1)
# never mask frames from the server to the client
b2 = 0
# How long is our payload?
length = len(payload)
if length < 126:
b2 |= length
message += chr(b2)
elif length < (2 ** 16) - 1:
b2 |= 126
message += chr(b2)
l = struct.pack(">H", length)
message += l
l = struct.pack(">Q", length)
b2 |= 127
message += chr(b2)
message += l
# Append payload to message
message += payload
# Send to the client
# Stolen from
def decodeCharArray(self, stringStreamIn):
# Turn string values into opererable numeric byte values
byteArray = [ord(character) for character in stringStreamIn]
datalength = byteArray[1] & 127
indexFirstMask = 2
if datalength == 126:
indexFirstMask = 4
elif datalength == 127:
indexFirstMask = 10
# Extract masks
masks = [m for m in byteArray[indexFirstMask : indexFirstMask+4]]
indexFirstDataByte = indexFirstMask + 4
# List of decoded characters
decodedChars = []
i = indexFirstDataByte
j = 0
# Loop through each byte that was received
while i < len(byteArray):
# Unmask this byte and add to the decoded buffer
decodedChars.append( chr(byteArray[i] ^ masks[j % 4]) )
i += 1
j += 1
# Return the decoded string
return decodedChars
# Handshake with this client
def dohandshake(self, header, key=None):
logging.debug("Begin handshake: %s" % header)
# Get the handshake template
handshake = self.handshake
# Step through each header
for line in header.split('\r\n')[1:]:
name, value = line.split(': ', 1)
# If this is the key
if name.lower() == "sec-websocket-key":
# Append the standard GUID and get digest
combined = value + MAGIC-GUID
response = base64.b64encode(hashlib.sha1(combined).digest())
# Replace the placeholder in the handshake response
handshake = handshake % { 'acceptstring' : response }
logging.debug("Sending handshake %s" % handshake)
return True
def onmessage(self, data):"Got message: %s" % data)
def send(self, data):"Sent message: %s" % data)
self.client.send("\x00%s\xff" % data)
def close(self):
# WebSocket server implementation
class WebSocketServer(object):
# Constructor
def __init__(self, bind, port, cls):
self.socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
self.socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
self.socket.bind((bind, port))
self.bind = bind
self.port = port
self.cls = cls
self.connections = {}
self.listeners = [self.socket]
# Listen for requests
def listen(self, backlog=5):
self.socket.listen(backlog)"Listening on %s" % self.port)
# Keep serving requests
self.running = True
while self.running:
# Find clients that need servicing
rList, wList, xList = select(self.listeners, [], self.listeners, 1)
for ready in rList:
if ready == self.socket:
logging.debug("New client connection")
client, address = self.socket.accept()
fileno = client.fileno()
self.connections[fileno] = self.cls(client, self)
logging.debug("Client ready for reading %s" % ready)
client = self.connections[ready].client
data = client.recv(4096)
fileno = client.fileno()
if data:
logging.debug("Closing client %s" % ready)
del self.connections[fileno]
# Step though and delete broken connections
for failed in xList:
if failed == self.socket:
logging.error("Socket broke")
for fileno, conn in self.connections:
self.running = False
# Entry point
if __name__ == "__main__":
logging.basicConfig(level=logging.DEBUG, format="%(asctime)s - %(levelname)s - %(message)s")
server = WebSocketServer("", 8000, WebSocket)
server_thread = Thread(target=server.listen, args=[5])
# Add SIGINT handler for killing the threads
def signal_handler(signal, frame):"Caught Ctrl+C, shutting down...")
server.running = False
signal.signal(signal.SIGINT, signal_handler)
while True:
Copy link

ddurdin commented Jan 25, 2013

Ran this on OSX 10.7 without changes. Worked great. Good piece of work.

Copy link

kanehekili commented May 9, 2013

Thanks! After a tedious work looking for a small working Websocket server I finally found yours. Runs perfectly on a Linux 64 machine.

Copy link

crossproduct commented Aug 29, 2013


Copy link

iloveluce commented Jun 25, 2014

Thanks for the gist, it's awesome. Just one thing, it does not correctly handle RST packets; when one of this is passed down it throws up a "error: [Errno 54] Connection reset by peer" error that causes the thread to crash. You can experience this by refreshing a website that has a WebSocket connection open

You can fix this by changing line 237 from

data = client.recv(4096)


    data = client.recv(4096)
except socket.error, e:
    logging.error("Connection error: %s" % e)
    data = None
    print "Connection error: %s" % e

Then it will throw the error but it won't cause the thread to crash. Thanks again!

Copy link

shennian commented Mar 12, 2015

It is awesome,but I can not get html files ,like this "localhost/test.html".
How should i do?
Thank you~

Copy link

sahilalipuria commented May 17, 2016

Can u send me the HTML files for this on

Copy link

arihernawan commented Jun 11, 2016

for someone who experience this error "SyntaxError: can't assign to operator", change MAGIC-GUID to MAGICGUID, remove strip from variable name.

Copy link

vdv7 commented Oct 18, 2016

doesn't work in python 3

Copy link

mog01 commented Mar 9, 2017

If I understand correctly, this solution is unable to handle partially received websocket messages? Or receiving multiple websocket messages within the same TCP read?

Copy link

ajavierfc commented Mar 19, 2018

Thank you

Copy link

deniz-yildirim06 commented Feb 7, 2019

Hi, How Ican get IP address of client?

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment