Websockets

Warning

Websockets are considered an experimental API. You’ll need to opt-in to this feature using the WEBSOCKETS feature flag:

app = Chalice('myapp')
app.experimental_feature_flags.update([
    'WEBSOCKETS'
])

See Experimental APIs for more information.

Chalice supports websockets through integration with an API Gateway Websocket API. If any of the decorators are present in a Chalice app, then an API Gateway Websocket API will be deployed and wired to Lambda Functions.

Responding to websocket events

In a Chalice app the websocket API is accessed through the three decorators on_ws_connect, on_ws_message, on_ws_disconnect. These handle a new websocket connection, an incoming message on an existing connection, and a connection being cleaned up respectively.

A decorated websocket handler function takes one argument event with the type WebsocketEvent. This class allows easy access to information about the API Gateway Websocket API, and information about the particular socket the handler is being invoked to serve.

Below is a simple working example application that prints to CloudWatch Logs for each of the events.

from boto3.session import Session
from chalice import Chalice

app = Chalice(app_name='test-websockets')
app.experimental_feature_flags.update([
    'WEBSOCKETS',
])
app.websocket_api.session = Session()


@app.on_ws_connect()
def connect(event):
    print('New connection: %s' % event.connection_id)


@app.on_ws_message()
def message(event):
    print('%s: %s' % (event.connection_id, event.body))


@app.on_ws_disconnect()
def disconnect(event):
    print('%s disconnected' % event.connection_id)

Sending a message over a websocket

To send a message to a websocket client Chalice, use the app.websocket_api.send() method. This method will work in any of the decorated functions outlined in the above section.

Two pieces of information are needed to send a message. The identifier of the websocket, and the contents for the message. Below is a simple example that when it receives a message, it sends back the message "I got your message!" over the same socket.

from boto3.session import Session
from chalice import Chalice

app = Chalice(app_name='test-websockets')
app.experimental_feature_flags.update([
    'WEBSOCKETS',
])
app.websocket_api.session = Session()


@app.on_ws_message()
def message(event):
    app.websocket_api.send(event.connection_id, 'I got your message!')

See Websocket Tutorials for completely worked example applications.

Continuous Deployment (CD) →