Tutorial (WebSockets)#

In this tutorial, we’re going to build a WebSocket server using Falcon. We’ll start with a simple server that echoes back any message it receives.

We’ll then add more functionality to the server, such as sending JSON data and logging messages.

Note

This tutorial covers the asynchronous flavor of Falcon using the ASGI protocol.

A Falcon WebSocket server builds upon the ASGI WebSocket specification. Therefore it’s not supported in a Falcon WSGI application.

First Steps#

We’ll start with a clean working directory and create a new virtual environment using the venv module:

$ mkdir ws_tutorial
$ cd ws_tutorial
$ python3 -m venv .venv
$ source .venv/bin/activate

Create the following directory structure:

ws_tutorial
├── .venv
└── ws_tutorial
    ├── __init__.py
    └── app.py

And next we’ll install Falcon and Uvicorn in our freshly created virtual environment:

$ pip install falcon uvicorn

Now, let’s create a simple Falcon application to ensure our project is working as expected.

import falcon.asgi
import uvicorn

app = falcon.asgi.App()

class HelloWorldResource:
    async def on_get(self, req, resp):
        resp.media = {'hello': 'world'}

app.add_route('/hello', HelloWorldResource())

if __name__ == '__main__':
    uvicorn.run(app, host='localhost', port=8000)

Now we can test the application with httpie (installable with pip install httpie) by running the following command:

$ http localhost:8000/hello

 HTTP/1.1 200 OK
 content-length: 18
 content-type: application/json
 date: Sat, 13 Jul 2024 09:13:24 GMT
 server: uvicorn

 {
     "hello": "world"
 }

Awesome, it works! Now let’s move on to building our WebSocket server.

WebSockets Server#

We will update our server to include a websocket route that will echo back any message it receives. Later we’ll update the server with more logic, but for now, let’s keep it simple.

import falcon.asgi
from falcon import WebSocketDisconnected
from falcon.asgi import Request, WebSocket
import uvicorn

app = falcon.asgi.App()


class HelloWorldResource:
    async def on_get(self, req, resp):
        resp.media = {'hello': 'world'}


class EchoWebSocketResource:
    async def on_websocket(self, req: Request, ws: WebSocket):
        try:
            await ws.accept()
        except WebSocketDisconnected:
            return

        while True:
            try:
                message = await ws.receive_text()
                await ws.send_text(f"Received the following text: {message}")
            except WebSocketDisconnected:
                return


app.add_route('/hello', HelloWorldResource())
app.add_route('/echo', EchoWebSocketResource())

if __name__ == '__main__':
    uvicorn.run(app, host='localhost', port=8000)

We’ll also need to install a websockets library. There are multiple ways to do this:

$ pip install websockets
or
$ pip install uvicorn[standard]
or
$ wsproto

To test the new WebSocket route, we can use the websocat tool:

$ websocat ws://localhost:8000/echo
$ hello
Received the following text: hello

Cool! We have a working WebSocket server. Now let’s add some more functionality to our server.

To make this easier, we’ll create a simple client that will send messages to our server.

Simple Client#

Create a new file called client.py in the same directory as app.py. The client will ask for your input and send it to the server.:

# This is a simple example of a WebSocket client that sends a message to the server.
# Since it's an example using the `websockets` library, and it isn't using anything
# specific to Falcon, there are no tests. Coverage is skipped for this module.

import asyncio

import websockets


async def send_message():
    uri = 'ws://localhost:8000/echo'

    async with websockets.connect(uri) as websocket:
        while True:
            message = input('Enter a message (q to exit): ')
            if message.casefold() == 'q':
                break
            await websocket.send(message)
            response = await websocket.recv()
            print(response)


if __name__ == '__main__':
    asyncio.run(send_message())

Run this client in a separate terminal:

$ python client.py
Enter a message: Hi
Received the following text: Hi

This will simplify testing our server.

Now let’s add some more functionality to our server.

We’ve been working with text input/output - let’s try sending sending some JSON data.

from datetime import datetime

import falcon.asgi
from falcon import WebSocketDisconnected
from falcon.asgi import Request, WebSocket
import uvicorn

app = falcon.asgi.App()


class HelloWorldResource:
    async def on_get(self, req, resp):
        resp.media = {'hello': 'world'}


class EchoWebSocketResource:
    async def on_websocket(self, req: Request, ws: WebSocket):
        try:
            await ws.accept()
        except WebSocketDisconnected:
            return

        while True:
            try:
                message = await ws.receive_text()
                await ws.send_media({'message': message, 'date': datetime.now().isoformat()})
            except WebSocketDisconnected:
                return


app.add_route('/hello', HelloWorldResource())
app.add_route('/echo', EchoWebSocketResource())

if __name__ == '__main__':
    uvicorn.run(app, host='localhost', port=8000)
$ python client.py
$ Enter a message: Hi
  {"message": "Hi", "date": "2024-07-13T12:11:51.758923"}

Note

By default, send_media() and receive_media() will serialize to (and deserialize from) JSON for a TEXT payload, and to/from MessagePack for a BINARY payload (see also: Built-in Media Handlers).

Lets try to query for data from the server. We’ll create a new resource that will return a report based on the query.

Server side:

from datetime import datetime

import falcon.asgi
from falcon import WebSocketDisconnected
from falcon.asgi import Request, WebSocket
import uvicorn

REPORTS = {
    'report1': {
        'title': 'Report 1',
        'content': 'This is the content of report 1',
    },
    'report2': {
        'title': 'Report 2',
        'content': 'This is the content of report 2',
    },
    'report3': {
        'title': 'Report 3',
        'content': 'This is the content of report 3',
    },
    'report4': {
        'title': 'Report 4',
        'content': 'This is the content of report 4',
    },
}

app = falcon.asgi.App()


class HelloWorldResource:
    async def on_get(self, req, resp):
        resp.media = {'hello': 'world'}


class EchoWebSocketResource:
    async def on_websocket(self, req: Request, ws: WebSocket):
        try:
            await ws.accept()
        except WebSocketDisconnected:
            return

        while True:
            try:
                message = await ws.receive_text()
                await ws.send_media({'message': message, 'date': datetime.now().isoformat()})
            except WebSocketDisconnected:
                return

class ReportsResource:
    async def on_websocket(self, req: Request, ws: WebSocket):
        try:
            await ws.accept()
        except WebSocketDisconnected:
            return

        while True:
            try:
                query = await ws.receive_text()
                report = REPORTS.get(query, None)
                print(report)

                if report is None:
                    await ws.send_media({'error': 'report not found'})
                    continue

                await ws.send_media({'report': report["title"]})
            except WebSocketDisconnected:
                return


app.add_route('/hello', HelloWorldResource())
app.add_route('/echo', EchoWebSocketResource())
app.add_route('/reports', ReportsResource())


if __name__ == '__main__':
    uvicorn.run(app, host='localhost', port=8000)

We’ll also create new client app (reports_client.py), that will connect to the reports endpoint. :

import asyncio
import websockets


async def send_message():
    uri = "ws://localhost:8000/reports"
    async with websockets.connect(uri) as websocket:
        while True:
            message = input("Name of the log: ")
            await websocket.send(message)
            response = await websocket.recv()
            print(response)


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

We’ve added a new resource that will return a report based on the query. The client will send a query to the server, and the server will respond with the report. If it can’t find the report, it will respond with an error message.

This is a simple example, but you can easily extend it to include more complex logic like fetching data from a database.

Middleware#

Falcon supports middleware, which can be used to add functionality to the application. For example, we can add a middleware that prints when a connection is established.

from datetime import datetime

import falcon.asgi
from falcon import WebSocketDisconnected
from falcon.asgi import Request, WebSocket
import uvicorn

REPORTS = {
    'report1': {
        'title': 'Report 1',
        'content': 'This is the content of report 1',
    },
    'report2': {
        'title': 'Report 2',
        'content': 'This is the content of report 2',
    },
    'report3': {
        'title': 'Report 3',
        'content': 'This is the content of report 3',
    },
    'report4': {
        'title': 'Report 4',
        'content': 'This is the content of report 4',
    },
}

app = falcon.asgi.App()

class LoggerMiddleware:
    async def process_request_ws(self, req: Request, ws: WebSocket):
        # This will be called for the HTTP request that initiates the
        #   WebSocket handshake before routing.
        pass

    async def process_resource_ws(self, req: Request, ws: WebSocket, resource, params):
        # This will be called for the HTTP request that initiates the
        #   WebSocket handshake after routing (if a route matches the
        #   request).
        print(f'WebSocket connection established on {req.path}')


class HelloWorldResource:
    async def on_get(self, req, resp):
        resp.media = {'hello': 'world'}


class EchoWebSocketResource:
    async def on_websocket(self, req: Request, ws: WebSocket):
        try:
            await ws.accept()
        except WebSocketDisconnected:
            return

        while True:
            try:
                message = await ws.receive_text()
                await ws.send_media({'message': message, 'date': datetime.now().isoformat()})
            except WebSocketDisconnected:
                return

class ReportsResource:
    async def on_websocket(self, req: Request, ws: WebSocket):
        try:
            await ws.accept()
        except WebSocketDisconnected:
            return

        while True:
            try:
                query = await ws.receive_text()
                report = REPORTS.get(query, None)
                print(report)

                if report is None:
                    await ws.send_media({'error': 'report not found'})
                    continue

                await ws.send_media({'report': report["title"]})
            except WebSocketDisconnected:
                return


app.add_route('/hello', HelloWorldResource())
app.add_route('/echo', EchoWebSocketResource())
app.add_route('/reports', ReportsResource())

app.add_middleware(LoggerMiddleware())


if __name__ == '__main__':
    uvicorn.run(app, host='localhost', port=8000)

Now, when you run the server, you should see a message in the console when a WebSocket connection is established.

Authentication#

Adding authentication can be done with the help of middleware as well. Authentication can be done a few ways. In this example we’ll use the First message method, as described on the websockets documentation.

There are some considerations to take into account when implementing authentication in a WebSocket server.

Updated server code:

from datetime import datetime
import logging
import pathlib

import uvicorn

from falcon import WebSocketDisconnected
import falcon.asgi
from falcon.asgi import Request
from falcon.asgi import WebSocket

logger = logging.getLogger('ws-logger')
logger.setLevel('INFO')
logger.addHandler(logging.StreamHandler())


REPORTS = {
    'report1': {
        'title': 'Report 1',
        'content': 'This is the content of report 1',
    },
    'report2': {
        'title': 'Report 2',
        'content': 'This is the content of report 2',
    },
    'report3': {
        'title': 'Report 3',
        'content': 'This is the content of report 3',
    },
    'report4': {
        'title': 'Report 4',
        'content': 'This is the content of report 4',
    },
}

app = falcon.asgi.App()


class LoggerMiddleware:
    async def process_request_ws(self, req: Request, ws: WebSocket):
        # This will be called for the HTTP request that initiates the
        #   WebSocket handshake before routing.
        pass

    async def process_resource_ws(self, req: Request, ws: WebSocket, resource, params):
        # This will be called for the HTTP request that initiates the
        #   WebSocket handshake after routing (if a route matches the
        #   request).
        logger.info('WebSocket connection established on %r', req.path)


class AuthMiddleware:
    def __init__(self, protected_routes: list[str] | None = None):
        if protected_routes is None:
            protected_routes = []

        self.protected_routes = protected_routes

    async def process_request_ws(self, req: Request, ws: WebSocket):
        # Opening a connection so we can receive the token
        await ws.accept()

        # Check if the route is protected
        if req.path not in self.protected_routes:
            return

        token = await ws.receive_text()

        if token != 'very secure token':
            await ws.close(1008)
            return

        # Never log tokens in production
        logger.info('Client with token %r Authenticated', token)


class HelloWorldResource:
    async def on_get(self, req, resp):
        resp.media = {'hello': 'world'}


class EchoWebSocketResource:
    async def on_websocket(self, req: Request, ws: WebSocket):
        while True:
            try:
                message = await ws.receive_text()
                await ws.send_media(
                    {'message': message, 'date': datetime.now().isoformat()}
                )
            except WebSocketDisconnected:
                return


class ReportsResource:
    async def on_websocket(self, req: Request, ws: WebSocket):
        while True:
            try:
                query = await ws.receive_text()
                report = REPORTS.get(query, None)
                logger.info('selected report: %s', report)

                if report is None:
                    await ws.send_media({'error': 'report not found'})
                    continue

                await ws.send_media({'report': report['title']})
            except WebSocketDisconnected:
                return


app.add_route('/hello', HelloWorldResource())
app.add_route('/echo', EchoWebSocketResource())
app.add_route('/reports', ReportsResource())

app.add_middleware(LoggerMiddleware())
app.add_middleware(AuthMiddleware(['/reports']))

# usually a web server, like Nginx or Caddy, should serve static assets, but
# for the purpose of this example we use falcon.
static_path = pathlib.Path(__file__).parent / 'static'
app.add_static_route('/', static_path, fallback_filename='index.html')

if __name__ == '__main__':
    uvicorn.run(app, host='localhost', port=8000)  # pragma: no cover

Updated client code for the reports client:

# This is a simple example of a WebSocket client that sends a message to the server.
# Since it's an example using the `websockets` library and it isn't using anything
# specific to Falcon, there are no tests. Coverage is skipped for this module.

import asyncio

import websockets


async def send_message():
    uri = 'ws://localhost:8000/reports'

    async with websockets.connect(uri) as websocket:
        # Send the authentication token
        await websocket.send('very secure token')

        while True:
            message = input('Name of the log (q to exit): ')
            if message.casefold() == 'q':
                break
            await websocket.send(message)
            response = await websocket.recv()
            print(response)


if __name__ == '__main__':
    asyncio.run(send_message())

Things we’ve changed:

  • Added a new middleware class AuthMiddleware that will check the token on the first message.

  • Opening a WebSocket connection is now handled by the middleware.

  • The client now sends a token as the first message, if required for that route.

  • Falcon was configured to serve a simple HTML page to use the echo WebSocket client for a browser.

If you try to query the reports endpoint now, everything works as expected on an authenticated route. But as soon as you remove/modify the token, the connection will be closed (after sending the first query - a downside of first-message authentication).

$ python reports_client.py
[...]
websockets.exceptions.ConnectionClosedError: received 1008 (policy violation); then sent 1008 (policy violation)

Note

This is a simple example of how to add authentication to a WebSocket server. In a real-world application, you would want to use a more secure method of authentication, such as JWT tokens.

What Now#

This tutorial is just the beginning. You can extend the server with more complex logic. For example, you could add a database to store/retrieve the reports, or add more routes to the server.

For more information on websockets in Falcon, check out the WebSocket API.