WS Gateway Streaming

Last Updated: June 25, 2026

Overview

In addition to the HTTP webhook flow (start()), the Careflow library supports real-time data delivery over WebSocket via the Careflow WS Gateway — a WebSocket-to-Kafka bridge. This is provided by the bundled careflow-wsgateway-client SDK and exposed through careflow.stream().

Function	Transport	Use Case
`start(callback)`	HTTP webhook (Data Broker)	Standard / batch workloads
`stream(callback)`	WebSocket (WS Gateway)	Real-time / low-latency workloads

Both entry points use the same callback signature. The callback receives one argument: the parsed data payload from the message. As of Careflow Broker library 0.2.0.102, stream is exported from the careflow package, so from careflow import stream works.

Configuration

Add a ws_gateway section to config.yaml:

yaml

ws_gateway:
 BASE_URL: 'wss://your-ws-gateway-host/ws'
 DEV_BASE_URL: 'ws://dev-gateway-host/ws'
 CLIENT_SECRET: 'client-secret'
 DEV_CLIENT_SECRET: 'dev-secret'
 SEEK_TO_LATEST: false

These fields are optional in the schema, but the pair required at runtime depends on the environment:

Environment	Behaviour	Required Fields
`local`	No WebSocket connection — falls back to the local sample-data timer (identical to `start()` in `local`)	None (uses sample feed)
`development`	Connects to the dev WS Gateway	`DEV_BASE_URL, DEV_CLIENT_SECRET`
Any other	Connects to the production WS Gateway	`BASE_URL, CLIENT_SECRET`

If the required fields for the active environment are missing, stream() logs an error and exits the process.

The subscribed topics are taken from dbroker.SUB_TOPIC (split on |). Messages on dbroker.CONTROL_TOPIC (if configured) are routed to the control_callback. Set ws_gateway.SEEK_TO_LATEST: true, or pass seek_to_latest=True to stream(), to connect at the latest offset instead of consuming backlog.

When deployed via the Careflow Orchestrator, the ws_gateway fields are populated automatically at deployment time.

Usage

python

import asyncio
from careflow import stream, logger

def callback(data):
  logger.info(f"Received data: {data}")
  # process inference input; publish results via careflow.send(...) or write to S3

if __name__ == '__main__':
    asyncio.run(stream(callback))

stream(callback, sample_parser=None, control_callback=None, seek_to_latest=None):

stream() is an async coroutine and must be awaited, for example with asyncio.run(...). Its parameters mirror start():

callback: single function (all topics) or a list mapped by SUB_TOPIC order
sample_parser: template values used only in local mode (sample fallback)
control_callback: invoked with messages on CONTROL_TOPIC
seek_to_latest: optional boolean override for ws_gateway.SEEK_TO_LATEST

The library creates and manages the WSGatewayClient internally, including connection, message routing, and teardown.

Reconnection Behaviour

The WS Gateway client reconnects automatically on unexpected connection loss using exponential backoff (initial 1 s, capped at 30 s). It will not reconnect when:

Authentication fails (invalid CLIENT_SECRET)
An unauthorized topic is requested
disconnect() is called explicitly

Connection lifecycle events (connect, disconnect, reconnect, error) are logged via careflow.logger.