Working with WebSocket APIs in Stelvio

This guide explains how to build WebSocket APIs with Stelvio using Amazon API Gateway WebSocket APIs. You'll learn how to define routes, protect $connect with authorizers, put the API behind a custom domain, and reply to clients with the management API.

WebsocketApi creates an API Gateway v2 WebSocket API, an auto-deploy stage (default $default), a CloudWatch log group for access logs, a Lambda function and integration for each handler, and the IAM permissions API Gateway needs to invoke those functions. Route selection defaults to $request.body.action: a JSON message {"action": "ping", ...} matches a ping route.

WebSocket APIs are for server push and two-way traffic: chat, live updates, and long-running jobs. For request/response HTTP endpoints, use HttpApi.

Creating a WebSocket API

from stelvio.aws.api_gateway import WebsocketApi

api = WebsocketApi("chat")
api.route("$connect", "functions/chat.connect")
api.route("$disconnect", "functions/chat.disconnect")
api.route("$default", "functions/chat.default")
api.route("ping", "functions/chat.ping")

That's enough for a working API. Stelvio uses the name for the AWS resource names and the console label.

The deployed endpoint is available as api.url and always uses the wss:// scheme. Without a custom domain it includes the stage path, including $default (HTTP APIs omit that stage from the URL):

api.url  # Output[str], resolves to wss://{api-id}.execute-api.{region}.amazonaws.com/$default

API Configuration

from stelvio.aws.api_gateway import ApiDomain, WebsocketApi

# Basic API with default settings
api = WebsocketApi("chat")

# Named stage instead of $default
api = WebsocketApi("chat", stage_name="production")

# Custom route selection (messages use {"type": "ping", ...})
api = WebsocketApi("chat", route_selection_expression="$request.body.type")

# API with a custom domain owned by this API
api = WebsocketApi("chat", domain_name="chat.example.com")

# API with a shared custom domain
domain = ApiDomain("public-domain", domain_name="api.example.com")
api = WebsocketApi("chat", domain=domain, api_mapping_key="chat")

# Disable the default execute-api hostname (requires a custom domain)
api = WebsocketApi(
    "chat",
    domain_name="chat.example.com",
    disable_execute_api_endpoint=True,
)

Available configuration options:

Defining Routes

api.route(route_key, handler)

Route keys use API Gateway's native values:

• $connect: runs when a client opens a connection. The return value is the handshake; a non-2xx status rejects the connection.
• $disconnect: runs when the connection closes. AWS does not guarantee it fires (the client can disappear or the network can drop).
• $default: catches messages that don't match another route
• Custom names such as ping or sendMessage: selected from $request.body.action

Each route key must be unique. Adding the same key twice raises an error.

api.route("$connect", "functions/chat.connect")
api.route("$default", "functions/chat.fallback")
api.route("sendMessage", "functions/chat.send")

Connecting Lambda Functions

The handler accepts the same forms as other Stelvio Lambda integrations: a path string, a FunctionConfig, a dictionary, or an existing Function instance.

Routes that point to the same handler share a single Lambda function and integration. Configure a shared function on only one of its routes:

api.route("$connect", "functions/chat.lifecycle", memory=512)
api.route("$disconnect", "functions/chat.lifecycle")

You can pass Lambda options directly when the handler is a string:

api.route(
    "sendMessage",
    "functions/chat.send",
    memory=512,
    timeout=20,
)

Lambda Event Format

WebSocket handlers receive the API Gateway WebSocket event. The connection ID and route key live under requestContext:

# functions/chat.py
import json

def connect(event, context):
    connection_id = event["requestContext"]["connectionId"]
    return {"statusCode": 200}  # not sent to the client

def send(event, context):
    route_key = event["requestContext"]["routeKey"]
    body = json.loads(event.get("body") or "{}")
    return {"statusCode": 200}  # not sent to the client

A client message like {"action": "sendMessage", "text": "hi"} invokes the sendMessage route; the JSON body is available as event["body"].

Authorization

Authorization runs only during $connect. Routes are public by default. Pass auth= on the $connect route to require a Lambda authorizer or IAM.

auth = api.add_lambda_authorizer(
    "token-auth",
    "functions/auth.authorize",
    identity_sources=["route.request.querystring.token"],
)
api.route("$connect", "functions/chat.connect", auth=auth)
api.route("$default", "functions/chat.default")

Passing auth= on any other route key raises an error.

Lambda Authorizers

WebSocket Lambda authorizers are REQUEST authorizers. Identity sources use WebSocket selection expressions such as route.request.header.Authorization and route.request.querystring.token.

The authorizer must return an API Gateway IAM policy response. WebSocket APIs do not support the HTTP API simple response format:

# functions/auth.py
def authorize(event, context):
    token = (event.get("queryStringParameters") or {}).get("token")
    effect = "Allow" if token == "allow" else "Deny"
    return {
        "principalId": "websocket-user",
        "policyDocument": {
            "Version": "2012-10-17",
            "Statement": [
                {
                    "Action": "execute-api:Invoke",
                    "Effect": effect,
                    "Resource": event["methodArn"],
                }
            ],
        },
    }

add_lambda_authorizer accepts:

• name: Unique authorizer name within the API
• handler: Lambda function path, config, or Function instance
• identity_sources: Non-empty list of selection expressions (required)
• **function_config: Additional Lambda configuration (memory, timeout, etc.)

IAM Authorization

For IAM authorization, clients must SigV4-sign the connection request:

api.route("$connect", "functions/chat.connect", auth="IAM")

Custom Domains

For a single API on a single domain, pass domain_name directly:

api = WebsocketApi("chat", domain_name="chat.example.com")

Custom domains require a DNS provider configured on your Stelvio app. See the DNS guide. Unless you pass certificate_arn on a shared ApiDomain, Stelvio creates the ACM certificate, validates it with DNS, creates the API Gateway domain, and publishes the DNS record.

Pass a domain component when multiple APIs should share one domain. Each API on the shared domain must use a distinct api_mapping_key.

from stelvio.aws.api_gateway import ApiDomain, WebsocketApi

domain = ApiDomain("public-domain", domain_name="api.example.com")

chat_api = WebsocketApi("chat", domain=domain, api_mapping_key="chat")
admin_api = WebsocketApi("admin-ws", domain=domain, api_mapping_key="admin")

With a custom domain, api.url uses wss:// and includes the mapping key when set, for example wss://api.example.com/chat.

To use an existing ACM certificate (for example a wildcard already in the account), pass certificate_arn on ApiDomain:

domain = ApiDomain(
    "public-domain",
    domain_name="api.example.com",
    certificate_arn="arn:aws:acm:us-east-1:123456789012:certificate/abc123",
)

Access Logs

WebsocketApi enables access logging by default with a 30-day retention. You can change the retention or keep logs indefinitely:

# Keep logs for 90 days
api = WebsocketApi("chat", access_log_retention_days=90)

# Keep logs indefinitely
api = WebsocketApi("chat", access_log_retention_days="forever")

Logs are written in JSON and include request ID, source IP, request time, route key, connection ID, event type, status, and any integration error message.

Linking

Link a WebsocketApi to a function when that function needs to call PostToConnection or read the API URL. Linking grants execute-api:ManageConnections and injects the API URL, management URL, and execution ARN.

A route handler can link to the same API:

from stelvio.aws.api_gateway import WebsocketApi

api = WebsocketApi("chat")
api.route("$connect", "functions/chat.connect")
api.route("ping", "functions/chat.reply", links=[api])

For an API named chat, the linked function receives these properties:

Link Permissions

Linked functions receive:

• execute-api:ManageConnections on {execution_arn}/*/*/@connections/*

Sending Messages to Clients

Use the linked management URL as the apigatewaymanagementapi endpoint:

# functions/chat.py
import json

import boto3
from stlv_resources import Resources

def reply(event, context):
    client = boto3.client(
        "apigatewaymanagementapi",
        endpoint_url=Resources.chat.api_management_url,
    )

    connection_id = event["requestContext"]["connectionId"]
    payload = json.dumps({"echo": event.get("body")}).encode("utf-8")
    try:
        client.post_to_connection(ConnectionId=connection_id, Data=payload)
    except client.exceptions.GoneException:
        pass
    return {"statusCode": 200}

post_to_connection raises GoneException when the connection is already closed.

Customization

WebsocketApi and ApiDomain support the customize parameter for overriding properties on the underlying Pulumi resources. For an overview of how customization works, see the Customization guide.

Resource Keys

Next Steps

• Working with HTTP APIs - Request/response HTTP endpoints on API Gateway v2.
• Working with Lambda Functions - Learn how Lambda packaging and configuration work.
• Authentication with Cognito - User pools and identity pools for custom or IAM auth.
• Linking - Learn how links generate environment variables and permissions.
• DNS - Configure a DNS provider for custom domains.