Library for building WebSocket servers and clients in Python

I'm trying to implement a WebSocket server that, once it's ready to accept connections, starts generating data periodically. Whenever data is generated (e.g. every 2 seconds), I want to broadcast the data to all connected clients.

I looked at the documentation and could not find an example like that. There is an example where the WebSocket server sends a message periodically to connected clients, but each client has its own "interval" (i.e. all clients won't receive the message at the same time).

It would be great if the documentation had such an example. I'm coming from Node.js and this is pretty easy to do with the 'ws' npm package. I'm new to Python, so I hope my question is not ridiculously stupid.

From https://github.com/aaugustin/websockets/issues/77#issuecomment-152921813

it would also be nice to use pythons context manager syntax to guarantee client code can be robust without any need for checking on their end.

something like this:

def echo_handler(websocket):
    with websocket as ws:
        # as long as the socket is open do your thing
        while True:
            message = yield from ws.recv()
            yield from ws.send(message)

it's not necessarily pretty (it took 3 indents to write anything useful), but it'll take some responsibility off a client.

I'm not sold on this approach yet but I think it's interesting and deserves its own ticket.

Hi there,

I am thinking about the best way to handle a bunch of connections using the least amount of machines (2 to 4 beefy servers probably), and I wonder if folks here have insights. With http I've been using gunicorn which makes it very easy to use many cores, but I'm a bit lost with asyncio on how to use all my cores. My application will likely just push data into redis and do little CPU work, but if there are a bunch of connected clients cpu might become an issue.

I thought about running multiple servers on multiple ports, and use nginx who's serving on port 80 as a load-balancer + proxy_pass (is that even possible) to go to multiple daemon running the websockets app (something like that?).

And then I'd have multiple services (each one, on a given machine, running under systemd) / each proxy_pass talk to one.

start_server = websockets.serve(hello, 'localhost', 8000) # for 'proxy_pass A'
start_server = websockets.serve(hello, 'localhost', 8001) # for 'proxy_pass B'
start_server = websockets.serve(hello, 'localhost', 8002) # for 'proxy_pass C, etc...'

I know it's a very vague question and not really an issue, but in absence of a better place to discuss this ... here we go !

Thanks,

• Benjamin

Running a project that uses websockets 9.0 against mypy, throws:

error: Module has no attribute "connect"

for

import websockets
await websockects.connect(uri)

and

Module 'websockets' has no attribute 'connect'

for

from websockets import connect

Version 8.1 of websockets has no such issue. Using the latest mypy 0.812.

Hello,

I've noticed an issue that appears to have started with websockets version 10.0. I have a websocket server run by the AWS API gateway. I use the websockets library to connect with my python program. Under 9.1 there was no issue receiving messages, but I noticed that as of 10.0 I stopped receiving the messages the server continuously sends (verified through wscat).

When I run with the logger I'd observe the connection being made as expected:

async with websockets.connect(my_url) as websocket:

DEBUG, > Upgrade: websocket
DEBUG, > Connection: Upgrade
DEBUG, > Sec-WebSocket-Key: cSPm5HYtYSZU3BoudrY91w==
DEBUG, > Sec-WebSocket-Version: 13
DEBUG, > Sec-WebSocket-Extensions: permessage-deflate; server_max_window_bits=12; client_max_window_bits=12
DEBUG, > User-Agent: Python/3.7 websockets/10.0
DEBUG, < HTTP/1.1 101 Switching Protocols
DEBUG, < Date: Wed, 06 Oct 2021 17:57:43 GMT
DEBUG, < Connection: upgrade
DEBUG, < upgrade: websocket
DEBUG, < sec-websocket-accept: e9Ztme4JDr4o30+FPAUZwoJMZcE=
DEBUG, < sec-websocket-extensions: permessage-deflate;server_max_window_bits=12
DEBUG, = connection is OPEN

Note that the server does not reply with any compression options. I think this is valid if the server doesn't support compression.

I then wait on messages like so:

while True:
    msg = await websocket.recv()

Everything appears normal from a connection perspective, except messages should be flowing regularly instead of just default ping / pong style responses:

DEBUG, %% sending keepalive ping
DEBUG, > PING 3c 93 7f 61 [binary, 4 bytes]
DEBUG, < PONG 3c 93 7f 61 [binary, 4 bytes]

I discovered if I set compression=None in connect ie:

async with websockets.connect(my_url, compression=None) as websocket:

Then messages from my server flow as usual as they did in 9.1.

It seems like compression should be something AWS supports on their side and will also engage with their support. However, since things appear to work with wscat, I think there may also be an issue with the websockets library itself in supporting servers that do not support compression.

To help debugging, I'm hosting a websocket server through the API Gateway which outputs regular messages to any connected client and does not require any authentication. I won't guarantee it will stay up indefinitely (sorry to any future readers!)

To test, run: wscat -c wss://n36vwxc045.execute-api.us-east-2.amazonaws.com/test This should work and you'll see the output < {"thanks": "for investigating this"} at about 1/2 Hz.

If you use the following code with the websockets 10.0 library to read the messages:

import asyncio
import websockets
import socket

import logging

logging.basicConfig(
    format="%(message)s",
    level=logging.DEBUG,
)

def main() -> None:
    asyncio.get_event_loop().run_until_complete(
        listen_forever(
            "wss://n36vwxc045.execute-api.us-east-2.amazonaws.com/test"
        )
    )


async def listen_forever(
    url: str
) -> None:
    """
    Listen for a websocket message, reconnecting if something bad happens.

    Args:
      url (str): The URL to which to connect.

    """
    # outer loop restarted every time the connection fails
    while True:
        try:
            async with websockets.connect(
                url,
                # compression=None  # UNCOMMENT THIS LINE TO GET MESSAGES
            ) as websocket:
                print('Websocket connected, waiting for messages...')
                while True:
                    msg = await websocket.recv()
                    print(msg)
        except socket.gaierror as exp:
            print(exp)
            continue
        except ConnectionRefusedError as exp:
            print(exp)
            continue
        except websockets.exceptions.ConnectionClosedOK:
            print(
                'Server closed connection (2 hour timeout?), reconnecting.'
            )
            continue
        except websockets.exceptions.InvalidStatusCode as exp:
            print(
                'Server returned an InvalidStatusCode - {}. Reconnecting.'.format(
                    exp
                )
            )
            continue
        except asyncio.streams.IncompleteReadError as exp:
            print(
                'Server an IncompleteReadError - {}. Reconnecting.'.format(
                    exp
                )
            )
            continue
        except Exception as exp:
            print(
                'Unhandled exception - {}. Reconnecting.'.format(
                    exp
                )
            )
            continue


if __name__ == '__main__':
    main()

You'll observe no message output unless you uncomment line 35.

Thanks for all your great work on this project!

TL;DR: Sending many messages in a loop (via a loop of await ws.send(...) or via passing an async generator to send() once), does not release control, and disconnects clients due to unanswered pings.

The longer story: I have a server that generates large messages to send to clients. Sending those messages takes time (about a minute), both because it takes some time to generate the messages themselves (CPU-intensive), but also because the network is rather slow.

The problem that I've noticed, is that await ws.send(...) does not release control even once done. I also tried to pass an async generator to send, and didn't get control back also (see example code below).

The outcome from that is that websocket's pings are not getting answered, and the client disconnects.

Example code: Here's a simple server that sends 20 packets of data to the client. I use time.sleep(1) instead of asyncio.sleep(1) to simulate actual CPU-intensive processing.

import asyncio
import time
import websockets

async def handler(websocket, path):
    async for request in websocket:
        print("Got a request to send stuff")
        for x in range(20):
            print("Sending", x)
            await websocket.send(str(x))
            time.sleep(1)  # Simulate CPU-intensive operation
        print("Done sending")

async def heartbeat():
    while True:
        await asyncio.sleep(1)
        print("Heartbeat: ", time.time())

async def main():
    await websockets.serve(handler, "localhost", 6789)
    await heartbeat()

asyncio.run(main())

The output looked something like this:

Heartbeat:  1607622802.574514
Got a request to send stuff
Sending 0
Sending 1
<... No heartbeats here ...>
Sending 18
Sending 19
Done sending
Heartbeat:  1607622823.145667
Heartbeat:  1607622824.1507761

As you can see, there're no hearbeats at all while in the sending loop, even though we do await each websocket.send() individually.

Adding await asyncio.sleep(0) right after the send, as a patch, does somewhat fixes the problem. (I start to get heardbeats every few (~5) seconds)

I also tried passing an async iterator into ws.send() (since the documentation of send clearly states that in that case control will be released), but no luck. Here's the new handler() function:

async def handler(websocket, path):
    async def gen():
        for x in range(20):
            yield str(x)
            time.sleep(1)  # Simulate CPU-heavy operation

    async for request in websocket:
        print("Got a request to send stuff")
        await websocket.send(gen())
        print("Done sending")

This is actually worse, as I didn't have a good place to add an await asyncio.sleep(0) in.

I've tested with Websockets 8.1, and both python3.7 and python3.8.

Fatal read error on socket transport protocol: <asyncio.sslproto.SSLProtocol object at 0x1177ce940> transport: <_SelectorSocketTransport fd=47 read=polling write=<idle, bufsize=0>> Traceback (most recent call last): File "/var/containers/Bundle/Application/9F565D30-C54B-4210-902B-874D7A5AB814/Pythonista3.app/Frameworks/Py3Kit.framework/pylib/asyncio/selector_events.py", line 725, in _read_ready data = self._sock.recv(self.max_size) TimeoutError: [Errno 60] Operation timed out

In several projects of mines I do something like the following, in order to handle scenario that may happen with connection errors and reconnection attempts:

    async def listen_forever(self):
        while True:
        # outer loop restarted every time the connection fails
            try:
                async with websockets.connect(self.url) as ws:
                    while True:
                    # listener loop
                        try:
                            reply = await asyncio.wait_for(ws.recv(), timeout=***)
                        except (asyncio.TimeoutError, websockets.exceptions.ConnectionClosed):
                            try:
                                pong = await ws.ping()
                                await asyncio.wait_for(pong, timeout=self.ping_timeout)
                                logger.debug('Ping OK, keeping connection alive...')
                                continue
                            except:
                                await asyncio.sleep(self.sleep_time)
                                break  # inner loop
                        # do stuff with reply object
            except socket.gaierror:
                # log something
                continue
            except ConnectionRefusedError:
                # log something else
                continue

and I was wondering whether (1) this makes sense at all, and (2) there is any shortcut already provided in websockets to what seems quite a recurrent behavior (assuming that is correct!)

Hello Augustin, from version 10 occurs an error during closing browser tab connected to my websocket server. It occurs only in approximately 30% cases. But everything was ok in version 9.1.

Tested in Google Chrome 94.

Traceback:

closing handshake failed
Traceback (most recent call last):
  File "/usr/local/lib/python3.9/site-packages/websockets/legacy/server.py", line 232, in handler
    await self.close()
  File "/usr/local/lib/python3.9/site-packages/websockets/legacy/protocol.py", line 779, in close
    await asyncio.shield(self.close_connection_task)
  File "/usr/local/lib/python3.9/site-packages/websockets/legacy/protocol.py", line 1309, in close_connection
    self.transport.write_eof()
  File "/usr/local/lib/python3.9/asyncio/selector_events.py", line 972, in write_eof
    self._sock.shutdown(socket.SHUT_WR)
OSError: [Errno 107] Socket not connected

Websocket server simplified code:

import asyncio
import signal
import websockets
import sys


class Chat(object):
	async def process(self, websocket: websockets.WebSocketServerProtocol, path: str):
		async for event in websocket:
			print(event)
			sys.stdout.flush()

	__call__ = process


async def server(stop):
	async with websockets.serve(Chat(), "0.0.0.0", 8765):
		await stop


if __name__ == "__main__":
	loop = asyncio.get_event_loop()
	stop = loop.create_future()
	loop.add_signal_handler(signal.SIGTERM, stop.set_result, None)
	loop.run_until_complete(server(stop))

Can you help me, please?

Thanks a lot, Jaromir

I'm not sure how to fix this error:

  Traceback (most recent call last):
    File "/root/src/streamer/stocks_producer.py", line 44, in <module>
      main()
    File "/root/src/streamer/stocks_producer.py", line 40, in main
      streamer.start()
    File "/root/src/streamer/polygon_streamer.py", line 67, in start
      self.loop.run_until_complete(asyncio.gather(*tasks))
    File "uvloop/loop.pyx", line 1451, in uvloop.loop.Loop.run_until_complete
    File "/root/src/streamer/polygon_streamer.py", line 47, in connect
      raise Exception(error_message)
  Exception: resetting connection: ()
  103 - 2019-02-13 15:30:14,095 - streamer.polygon_streamer - ERROR - resetting connection: ('WebSocket connection is closed: code = 1006 (connection closed abnormally [internal]), no reason',)
  Traceback (most recent call last):
    File "/opt/conda/lib/python3.7/site-packages/websockets/protocol.py", line 528, in transfer_data
      msg = yield from self.read_message()
    File "/opt/conda/lib/python3.7/site-packages/websockets/protocol.py", line 580, in read_message
      frame = yield from self.read_data_frame(max_size=self.max_size)
    File "/opt/conda/lib/python3.7/site-packages/websockets/protocol.py", line 645, in read_data_frame
      frame = yield from self.read_frame(max_size)
    File "/opt/conda/lib/python3.7/site-packages/websockets/protocol.py", line 710, in read_frame
      extensions=self.extensions,
    File "/opt/conda/lib/python3.7/site-packages/websockets/framing.py", line 100, in read
      data = yield from reader(2)
    File "/opt/conda/lib/python3.7/asyncio/streams.py", line 677, in readexactly
      raise IncompleteReadError(incomplete, n)
  asyncio.streams.IncompleteReadError: 0 bytes read on a total of 2 expected bytes

here is my code snippet, thank you!

async def connect(self):
	while True:
		async with websockets.connect(self.url, max_queue=8 * 2**10,) as websocket:
			await websocket.send(json.dumps({"action": "auth", "params": self._api_key}))
			await websocket.send(json.dumps({"action": "subscribe", "params": self._symbols_str}))
			log.info("connected: {}".format(websocket.remote_address))
			while(True):
				try:
					message_str = await asyncio.wait_for(websocket.recv(), timeout=self._timeout)
					# await self.queue.put(message_str)
					self.queue.put_nowait(message_str)
				except asyncio.TimeoutError:
					log.warn("timeout error - no data in {} seconds, pinging connection".format(self._timeout))
					pong_waiter = await asyncio.wait_for(websocket.ping(data="keepalive"), timeout=self._timeout)
					await asyncio.wait_for(pong_waiter, timeout=2 * self._timeout)
					log.warn('ping/pong received, keeping connection alive...')
				except Exception as e:
					error_message = "resetting connection: {}".format(e.args)
					log.error(error_message)
					raise Exception(error_message)

Traceback (most recent call last): File "/srv/locix/locix/websocket/server.py", line 184, in ping_handler await websocket.ping() File "/usr/local/lib/python3.5/dist-packages/websockets/protocol.py", line 433, in ping yield from self.write_frame(OP_PING, data) File "/usr/local/lib/python3.5/dist-packages/websockets/protocol.py", line 654, in write_frame extensions=self.extensions, File "/usr/local/lib/python3.5/dist-packages/websockets/framing.py", line 210, in write writer(output.getvalue()) File "/usr/lib/python3.5/asyncio/streams.py", line 294, in write self._transport.write(data) File "/usr/lib/python3.5/asyncio/selector_events.py", line 687, in write raise RuntimeError('Cannot call write() after write_eof()')

Bumps pypa/cibuildwheel from 2.11.1 to 2.11.4.

Dependabot will resolve any conflicts with this PR as long as you don't alter it yourself. You can also trigger a rebase manually by commenting @dependabot rebase.

https://websockets.readthedocs.io/en/stable/faq/common.html#how-can-i-pass-arguments-to-a-custom-protocol-subclass The code did't work on Python 3.10 now.

Here is a example about the reson behind:

import functools
class a:
    def __init__(self,extra_argument,*args,**kwargs):
        print(extra_argument)
        print(args)
        print(kwargs)
A=functools.partial(a,extra_argument=1)
hello=A(2)

The output is:

Traceback (most recent call last):
  File "/home/***/test.py", line 8, in <module>
    hello=A(2)
TypeError: a.__init__() got multiple values for argument 'extra_argument'

Maybe this way is outdated?

[Any] means one argument of Any type, ... means "ignore". This is relevant since WebSocketServerProtocol and similar in fact have multiple arguments. Ideally a Protocol with a __call__ method included should be used, but that's up for discussion since that would add maintainability issues as the signature would be duplicated in multiple places.

Can we please publish a document or update in ReadMe about

How to make a choice between websockets and websocket-client

If we can get elaborated comments on the downsides and upsides of using one over the other, it will be very helpful.

Add CIFuzz workflow action to have fuzzers build and run on each PR.

This is a service offered by OSS-Fuzz where websockets was recently integrated. CIFuzz can help detect regressions and catch fuzzing build issues early, and has a variety of features (see the URL above). In the current PR the fuzzers gets build on a pull request and will run for 300 seconds.