WebSocket (ASGI Only)

Falcon builds upon the ASGI WebSocket Specification to provide a simple, no-nonsense WebSocket server implementation.

With support for both WebSocket and Server-Sent Events (SSE), Falcon facilitates real-time, event-oriented communication between an ASGI application and a web browser, mobile app, or other client application.

Note

See also falcon.asgi.Response.sse to learn more about Falcon’s Server-Sent Event (SSE) support.

Usage

With Falcon you can easily add WebSocket support to any route in your ASGI app, simply by implementing an on_websocket() responder in the resource class for that route. As with regular HTTP requests, WebSocket flows can be augmented with middleware components and media handlers.

When a WebSocket handshake arrives (via a standard HTTP request), Falcon will first route it as usual to a specific resource class instance. Along the way, the following middleware methods will be invoked, if implemented on any middleware objects configured for the app:

class SomeMiddleware:
    async def process_request_ws(self, req, ws):
        """Process a WebSocket handshake request before routing it.

        Note:
            Because Falcon routes each request based on req.path, a
            request can be effectively re-routed by setting that
            attribute to a new value from within process_request().

        Args:
            req: Request object that will eventually be
                passed into an on_websocket() responder method.
            ws: The WebSocket object that will be passed into
                on_websocket() after routing.
        """

    async def process_resource_ws(self, req, ws, resource, params):
        """Process a WebSocket handshake request after routing.

        Note:
            This method is only called when the request matches
            a route to a resource.

        Args:
            req: Request object that will be passed to the
                routed responder.
            ws: WebSocket object that will be passed to the
                routed responder.
            resource: Resource object to which the request was
                routed.
            params: A dict-like object representing any additional
                params derived from the route's URI template fields,
                that will be passed to the resource's responder
                method as keyword arguments.
        """

If a route is found for the requested path, the framework will then check for a responder coroutine named on_websocket() on the target resource. If the responder is found, it is invoked in a similar manner to a regular on_get() responder, except that a falcon.asgi.WebSocket object is passed in, instead of an object of type falcon.asgi.Response.

For example, given a route that includes an account_id path parameter, the framework would expect an on_websocket() responder similar to this:

async def on_websocket(self, req: Request, ws: WebSocket, account_id: str):
    pass

If no route matches the path requested in the WebSocket handshake, control then passes to a default responder that simply raises an instance of HTTPRouteNotFound. By default, this error will be rendered as a 403 response with a 3404 close code. This behavior can be modified by adding a custom error handler (see also: add_error_handler()).

Similarly, if a route exists but the target resource does not implement an on_websocket() responder, the framework invokes a default responder that raises an instance of HTTPMethodNotAllowed. This class will be rendered by default as a 403 response with a 3405 close code.

Lost Connections

When the app attempts to receive a message from the client, the ASGI server emits a disconnect event if the connection has been lost for any reason. Falcon surfaces this event by raising an instance of WebSocketDisconnected to the caller.

On the other hand, the ASGI spec requires the ASGI server to silently consume messages sent by the app after the connection has been lost (i.e., it should not be considered an error). Therefore, an endpoint that primarily streams outbound events to the client might continue consuming resources unnecessarily for some time after the connection is lost.

As a workaround, Falcon implements a small incoming message queue that is used to detect a lost connection and then raise an instance of WebSocketDisconnected to the caller the next time it attempts to send a message.

This workaround is only necessary when the app itself does not consume messages from the client often enough to quickly detect when the connection is lost. Otherwise, Falcon’s receive queue can be disabled for a slight performance boost by setting max_receive_queue to 0 via ws_options.

Note also that some ASGI server implementations do not strictly follow the ASGI spec in this regard, and in fact will raise an error when the app attempts to send a message after the client disconnects. If testing reveals this to be the case for your ASGI server of choice, Falcon’s own receive queue can be safely disabled.

Error Handling

Falcon handles errors raised by an on_websocket() responder in a similar way to errors raised by other responders, with the following caveats.

First, when calling a custom error handler, the framework will pass None for the resp argument, while the WebSocket object representing the current connection will be passed as a keyword argument named ws:

async def my_error_handler(req, resp, ex, params, ws=None):
    # When invoked as a result of an error being raised by an
    #   on_websocket() responder, resp will be None and
    #   ws will be the same falcon.asgi.WebSocket object that
    #   was passed into the responder.
    pass

Second, it’s important to note that if no route matches the path in the WebSocket handshake request, or the matched resource does not implement an on_websocket() responder, the default HTTP error responders will be invoked, resulting in the request being denied with an HTTP 403 response and a WebSocket close code of either 3404 (Not Found) or 3405 (Method Not Allowed). Generally speaking, if either a default responder or on_websocket() raises an instance of HTTPError, the default error handler will close the WebSocket connection with a framework close code derived by adding 3000 to the HTTP status code (e.g., 3404).

Finally, in the case of a generic unhandled exception, a default error handler is invoked that will do its best to clean up the connection, closing it with the standard WebSocket close code 1011 (Internal Error). If your ASGI server does not support this code, the framework will use code 3011 instead; or you can customize it via the error_close_code property of ws_options.

As with any responder, the default error handlers for the app may be overridden via add_error_handler().

Media Handlers

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).

Note

In order to use the default MessagePack handler, the extra msgpack package (version 0.5.2 or higher) must be installed in addition to falcon from PyPI:

$ pip install msgpack

WebSocket media handling can be customized by using falcon.asgi.App.ws_options to specify an alternative handler for one or both payload types, as in the following example.

# Let's say we want to use a faster JSON library. You could also use this
#   pattern to add serialization support for custom types that aren't
#   normally JSON-serializable out of the box.
class RapidJSONHandler(falcon.media.TextBaseHandlerWS):
    def serialize(self, media: object) -> str:
        return rapidjson.dumps(media, ensure_ascii=False)

    # The raw TEXT payload will be passed as a Unicode string
    def deserialize(self, payload: str) -> object:
        return rapidjson.loads(payload)


# And/or for binary mode we want to use CBOR:
class CBORHandler(media.BinaryBaseHandlerWS):
    def serialize(self, media: object) -> bytes:
        return cbor2.dumps(media)

    # The raw BINARY payload will be passed as a byte string
    def deserialize(self, payload: bytes) -> object:
        return cbor2.loads(payload)

app = falcon.asgi.App()

# Expected to (de)serialize from/to str
json_handler = RapidJSONHandler()
app.ws_options.media_handlers[falcon.WebSocketPayloadType.TEXT] = json_handler

# Expected to (de)serialize from/to bytes, bytearray, or memoryview
cbor_handler = ProtocolBuffersHandler()
app.ws_options.media_handlers[falcon.WebSocketPayloadType.BINARY] = cbor_handler

The falcon module defines the following Enum values for specifying the WebSocket payload type:

falcon.WebSocketPayloadType.TEXT
falcon.WebSocketPayloadType.BINARY

Extended Example

Here is a more comprehensive (albeit rather contrived) example that illustrates some of the different ways an application can interact with a WebSocket connection. This example also introduces some common WebSocket errors raised by the framework.

import falcon.asgi
import falcon.media


class SomeResource:

    # Get a paginated list of events via a regular HTTP request.
    #
    #   For small-scale, all-in-one apps, it may make sense to support
    #   both a regular HTTP interface and one based on WebSocket
    #   side-by-side in the same deployment. However, these two
    #   interaction models have very different performance characteristics,
    #   and so larger scale-out deployments may wish to specifically
    #   designate instance groups for one type of traffic vs. the
    #   other (although the actual applications may still be capable
    #   of handling both modes).
    #
    async def on_get(self, req: Request, account_id: str):
        pass

    # Push event stream to client. Note that the framework will pass
    #   parameters defined in the URI template as with HTTP method
    #   responders.
    async def on_websocket(self, req: Request, ws: WebSocket, account_id: str):

        # The HTTP request used to initiate the WebSocket handshake can be
        #   examined as needed.
        some_header_value = req.get_header('Some-Header')

        # Reject it?
        if some_condition:
            # If close() is called before accept() the code kwarg is
            #   ignored, if present, and the server returns a 403
            #   HTTP response without upgrading the connection.
            await ws.close()
            return

        # Examine subprotocols advertised by the client. Here let's just
        #   assume we only support wamp, so if the client doesn't advertise
        #   it we reject the connection.
        if 'wamp' not in ws.subprotocols:
            # If close() is not called explicitly, the framework will
            #   take care of it automatically with the default code (1000).
            return

        # If, after examining the connection info, you would like to accept
        #   it, simply call accept() as follows:
        try:
            await ws.accept(subprotocol='wamp')
        except WebSocketDisconnected:
            return

        # Simply start sending messages to the client if this is an event
        #   feed endpoint.
        while True:
            try:
                event = await my_next_event()

                # Send an instance of str as a WebSocket TEXT (0x01) payload
                await ws.send_text(event)

                # Send an instance of bytes, bytearray, or memoryview as a
                #   WebSocket BINARY (0x02) payload.
                await ws.send_data(event)

                # Or if you want it to be serialized to JSON (by default; can
                #   be customized via app.ws_options.media_handlers):
                await ws.send_media(event)  # Defaults to WebSocketPayloadType.TEXT
            except WebSocketDisconnected:
                # Do any necessary cleanup, then bail out
                return

        # ...or loop like this to implement a simple request-response protocol
        while True:
            try:
                # Use this if you expect a WebSocket TEXT (0x01) payload,
                #   decoded from UTF-8 to a Unicode string.
                payload_str = await ws.receive_text()

                # Or if you are expecting a WebSocket BINARY (0x02) payload,
                #   in which case you will end up with a byte string result:
                payload_bytes = await ws.receive_data()

                # Or if you want to get a serialized media object (defaults to
                #   JSON deserialization of text payloads, and MessagePack
                #   deserialization for BINARY payloads, but this can be
                #   customized via app.ws_options.media_handlers).
                media_object = await ws.receive_media()

            except WebSocketDisconnected:
                # Do any necessary cleanup, then bail out
                return
            except TypeError:
                # The received message payload was not of the expected
                #   type (e.g., got BINARY when TEXT was expected).
                pass
            except json.JSONDecodeError:
                # The default media deserializer uses the json standard
                #   library, so you might see this error raised as well.
                pass

            # At any time, you may decide to close the websocket. If the
            #   socket is already closed, this call does nothing (it will
            #   not raise an error.)
            if we_are_so_done_with_this_conversation():
                # https://developer.mozilla.org/en-US/docs/Web/API/CloseEvent
                await ws.close(code=1000)
                return

            try:
                # Here we are sending as a binary (0x02) payload type, which
                #   will go find the handler configured for that (defaults to
                #   MessagePack which assumes you've also installed that
                #   package, but this can be customized as mentioned above.')
                await ws.send_media(
                    {'event': 'message'},
                    payload_type=WebSocketPayloadType.BINARY,
                )

            except WebSocketDisconnected:
                # Do any necessary cleanup, then bail out. If ws.close() was
                #   not already called by the app, the framework will take
                #   care of it.

                # NOTE: If you do not handle this exception, it will be
                #   bubbled up to a default error handler that simply
                #   logs the message as a warning and then closes the
                #   server side of the connection. This handler can be
                #   overridden as with any other error handler for the app.

                return

        # ...or run a couple of different loops in parallel to support
        #  independent bidirectional message streams.

        messages = collections.deque()

        async def sink():
            while True:
                try:
                    message = await ws.receive_text()
                except falcon.WebSocketDisconnected:
                    break

                messages.append(message)

        sink_task = falcon.create_task(sink())

        while not sink_task.done():
            while ws.ready and not messages and not sink_task.done():
                await asyncio.sleep(0)

            try:
                await ws.send_text(messages.popleft())
            except falcon.WebSocketDisconnected:
                break

        sink_task.cancel()
        try:
            await sink_task
        except asyncio.CancelledError:
            pass


class SomeMiddleware:
    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).
        pass


app = falcon.asgi.App(middleware=SomeMiddleware())
app.add_route('/{account_id}/messages', SomeResource())

Testing

Falcon’s testing framework includes support for simulating WebSocket connections with the falcon.testing.ASGIConductor class, as demonstrated in the following example.

# This context manages the ASGI app lifecycle, including lifespan events
async with testing.ASGIConductor(some_app) as c:
    async def post_events():
        for i in range(100):
            await c.simulate_post('/events', json={'id': i}):
            await asyncio.sleep(0.01)

    async def get_events_ws():
        # Simulate a WebSocket connection
        async with c.simulate_ws('/events') as ws:
            while some_condition:
                message = await ws.receive_text()

    asyncio.gather(post_events(), get_events_ws())

See also: simulate_ws().

Reference

WebSocket Class

The framework passes an instance of the following class into the on_websocket() responder. Conceptually, this class takes the place of the falcon.asgi.Response class for WebSocket connections.

class falcon.asgi.WebSocket(ver: str, scope: dict, receive: Callable[[], Awaitable[dict]], send: Callable[[dict], Awaitable], media_handlers: Mapping[WebSocketPayloadType, BinaryBaseHandlerWS | TextBaseHandlerWS], max_receive_queue: int)[source]

Represents a single WebSocket connection with a client.

ready

True if the WebSocket connection has been accepted and the client is still connected, False otherwise.

Type:

bool

unaccepted

True if the WebSocket connection has not yet been accepted, False otherwise.

Type:

bool)

closed

True if the WebSocket connection has been closed by the server or the client has disconnected.

Type:

bool

subprotocols

The list of subprotocol strings advertised by the client, or an empty tuple if no subprotocols were specified.

Type:

tuple[str]

supports_accept_headers

True if the ASGI server hosting the app supports sending headers when accepting the WebSocket connection, False otherwise.

Type:

bool

async accept(subprotocol: str | None = None, headers: Iterable[Iterable[str]] | Mapping[str, str] | None = None)[source]

Accept the incoming WebSocket connection.

If, after examining the connection’s attributes (headers, advertised subprotocols, etc.) the request should be accepted, the responder must first await this coroutine method to finalize the WebSocket handshake. Alternatively, the responder may deny the connection request by awaiting the close() method.

Keyword Arguments:
  • subprotocol (str) –

    The subprotocol the app wishes to accept, out of the list of protocols that the client suggested. If more than one of the suggested protocols is acceptable, the first one in the list from the client should be selected (see also: subprotocols).

    When left unspecified, a Sec-WebSocket-Protocol header will not be included in the response to the client. The client may choose to abandon the connection in this case, if it does not receive an explicit protocol selection.

  • headers (Iterable[[str, str]]) –

    An iterable of [name: str, value: str] two-item iterables, representing a collection of HTTP headers to include in the handshake response. Both name and value must be of type str and contain only US-ASCII characters.

    Alternatively, a dict-like object may be passed that implements an items() method.

    Note

    This argument is only supported for ASGI servers that implement spec version 2.1 or better. If an app needs to be compatible with multiple ASGI servers, it can reference the supports_accept_headers property to determine if the hosting server supports this feature.

async close(code: int | None = None) None[source]

Close the WebSocket connection.

This coroutine method sends a WebSocket CloseEvent to the client and then proceeds to actually close the connection.

The responder can also use this method to deny a connection request simply by awaiting it instead of accept(). In this case, the client will receive an HTTP 403 response to the handshake.

Keyword Arguments:

code (int) – The close code to use for the CloseEvent (default 1000). See also: https://developer.mozilla.org/en-US/docs/Web/API/CloseEvent

async receive_data() bytes[source]

Receive a message from the client with a binary data payload.

Awaiting this coroutine will block until a message is available or the WebSocket is disconnected.

async receive_media() object[source]

Receive a deserialized object from the client.

The incoming payload type determines the media handler that will be used to deserialize the object (see also: Media Handlers).

async receive_text() str[source]

Receive a message from the client with a Unicode string payload.

Awaiting this coroutine will block until a message is available or the WebSocket is disconnected.

async send_data(payload: bytes | bytearray | memoryview) None[source]

Send a message to the client with a binary data payload.

Parameters:

payload (Union[bytes, bytearray, memoryview]) – The binary data to send.

async send_media(media: object, payload_type: WebSocketPayloadType = WebSocketPayloadType.TEXT) None[source]

Send a serializable object to the client.

The payload type determines the media handler that will be used to serialize the given object (see also: Media Handlers).

Parameters:

media (object) – The object to send.

Keyword Arguments:

payload_type (falcon.WebSocketPayloadType) –

The payload type to use for the message (default falcon.WebSocketPayloadType.TEXT).

Must be one of:

falcon.WebSocketPayloadType.TEXT
falcon.WebSocketPayloadType.BINARY

async send_text(payload: str) None[source]

Send a message to the client with a Unicode string payload.

Parameters:

payload (str) – The string to send.

Built-in Media Handlers

class falcon.media.TextBaseHandlerWS[source]

Abstract Base Class for a WebSocket TEXT media handler.

deserialize(payload: str) object[source]

Deserialize TEXT payloads from a Unicode string.

By default, this method raises an instance of NotImplementedError. Therefore, it must be overridden if the child class wishes to support deserialization from TEXT (0x01) message payloads.

Parameters:

payload (str) – Message payload to deserialize.

Returns:

A deserialized object.

Return type:

object

serialize(media: object) str[source]

Serialize the media object to a Unicode string.

By default, this method raises an instance of NotImplementedError. Therefore, it must be overridden if the child class wishes to support serialization to TEXT (0x01) message payloads.

Parameters:

media (object) – A serializable object.

Returns:

The resulting serialized string from the input object.

Return type:

str

class falcon.media.BinaryBaseHandlerWS[source]

Abstract Base Class for a WebSocket BINARY media handler.

deserialize(payload: bytes) object[source]

Deserialize BINARY payloads from a byte string.

By default, this method raises an instance of NotImplementedError. Therefore, it must be overridden if the child class wishes to support deserialization from BINARY (0x02) message payloads.

Parameters:

payload (bytes) – Message payload to deserialize.

Returns:

A deserialized object.

Return type:

object

serialize(media: object) bytes | bytearray | memoryview[source]

Serialize the media object to a byte string.

By default, this method raises an instance of NotImplementedError. Therefore, it must be overridden if the child class wishes to support serialization to BINARY (0x02) message payloads.

Parameters:

media (object) – A serializable object.

Returns:

The resulting serialized byte string from the input object. May be an instance of bytes, bytearray, or memoryview.

Return type:

bytes

class falcon.media.JSONHandlerWS(dumps=None, loads=None)[source]

WebSocket media handler for de(serializing) JSON to/from TEXT payloads.

This handler uses Python’s standard json library by default, but can be easily configured to use any of a number of third-party JSON libraries, depending on your needs. For example, you can often realize a significant performance boost under CPython by using an alternative library. Good options in this respect include orjson, python-rapidjson, and mujson.

Note

If you are deploying to PyPy, we recommend sticking with the standard library’s JSON implementation, since it will be faster in most cases as compared to a third-party library.

Overriding the default JSON implementation is simply a matter of specifying the desired dumps and loads functions:

import falcon
from falcon import media

import rapidjson

json_handler = media.JSONHandlerWS(
    dumps=rapidjson.dumps,
    loads=rapidjson.loads,
)

app = falcon.asgi.App()
app.ws_options.media_handlers[falcon.WebSocketPayloadType.TEXT] = json_handler

By default, ensure_ascii is passed to the json.dumps function. If you override the dumps function, you will need to explicitly set ensure_ascii to False in order to enable the serialization of Unicode characters to UTF-8. This is easily done by using functools.partial to apply the desired keyword argument. In fact, you can use this same technique to customize any option supported by the dumps and loads functions:

from functools import partial

from falcon import media
import rapidjson

json_handler = media.JSONHandlerWS(
    dumps=partial(
        rapidjson.dumps,
        ensure_ascii=False, sort_keys=True
    ),
)
Keyword Arguments:
  • dumps (func) – Function to use when serializing JSON.

  • loads (func) – Function to use when deserializing JSON.

class falcon.media.MessagePackHandlerWS[source]

WebSocket media handler for de(serializing) MessagePack to/from BINARY payloads.

This handler uses msgpack.unpackb() and msgpack.packb(). The MessagePack bin type is used to distinguish between Unicode strings (of type str) and byte strings (of type bytes).

Note

This handler requires the extra msgpack package (version 0.5.2 or higher), which must be installed in addition to falcon from PyPI:

$ pip install msgpack

Error Types

class falcon.WebSocketDisconnected(code: int | None = None)[source]

The websocket connection is lost.

This error is raised when attempting to perform an operation on the WebSocket and it is determined that either the client has closed the connection, the server closed the connection, or the socket has otherwise been lost.

Keyword Arguments:

code (int) – The WebSocket close code, as per the WebSocket spec (default 1000).

code

The WebSocket close code, as per the WebSocket spec.

Type:

int

class falcon.WebSocketPathNotFound(code: int | None = None)[source]

No route could be found for the requested path.

A simulated WebSocket connection was attempted but the path specified in the handshake request did not match any of the app’s routes.

class falcon.WebSocketHandlerNotFound(code: int | None = None)[source]

The routed resource does not contain an on_websocket() handler.

class falcon.WebSocketServerError(code: int | None = None)[source]

The server encountered an unexpected error.

class falcon.PayloadTypeError[source]

The WebSocket message payload was not of the expected type.

Options

class falcon.asgi.WebSocketOptions[source]

Defines a set of configurable WebSocket options.

An instance of this class is exposed via falcon.asgi.App.ws_options for configuring certain WebSocket behaviors.

error_close_code

The WebSocket close code to use when an unhandled error is raised while handling a WebSocket connection (default 1011). For a list of valid close codes and ranges, see also: https://tools.ietf.org/html/rfc6455#section-7.4

Type:

int

media_handlers

A dict-like object for configuring media handlers according to the WebSocket payload type (TEXT vs. BINARY) of a given message. See also: Media Handlers.

Type:

dict

max_receive_queue

The maximum number of incoming messages to enqueue if the reception rate exceeds the consumption rate of the application (default 4). When this limit is reached, the framework will wait to accept new messages from the ASGI server until the application is able to catch up.

This limit applies to Falcon’s incoming message queue, and should generally be kept small since the ASGI server maintains its own receive queue. Falcon’s queue can be disabled altogether by setting max_receive_queue to 0 (see also: Lost Connections).

Type:

int