StreamMachine

Async stream processing on Redis Streams for Python. Register consumers and periodic producers with decorators, run them on one event loop, and scale out by starting more processes in the same consumer group.

from streammachine import App, Message

app = App(name="demo")

@app.timer(1)
async def producer():
    await app.send("greetings", {"message": "hello"})

@app.agent("greetings", group="greeters")
async def consumer(record: Message):
    print("received:", record.message)

if __name__ == "__main__":
    app.start()

Features

• Agents and timers: @app.agent(stream, group=...) consumes a stream through a Redis consumer group, @app.timer(seconds) runs a coroutine periodically.
• Resilient consumers: handler errors are logged and skipped; Redis outages reconnect with bounded exponential backoff under a stable consumer name so pending entries are never orphaned.
• Bounded streams: App(stream_maxlen=...) or STREAMMACHINE_STREAM_MAXLEN trims every produced stream with approximate MAXLEN.
• Shared state: app.storage is a multiprocessing.Manager backed key/value store with per-key async locks.
• DataFrames: streams_to_dataframe() and TimeSeriesBuffer turn raw XREAD output into pandas frames with automatic pruning.
• OHLC aggregation: create_ohlc_aggregator() builds candles from tick streams, with an optional Cython build for higher throughput.
• Optional extras: a FastAPI monitoring dashboard, an MCP server exposing streams and storage as tools, and pickle-based object storage.

Installation

pip install streammachine

Extras:

Requires Python 3.10+ and a Redis server (6.2 or newer recommended). uvloop is used automatically on Linux and macOS.

Configuration

Connection settings are read from the environment:

How it works

1. Decorators attach metadata to your handlers (via venusian); nothing runs at import time.
2. app.start() scans the calling module, creates one consumer task per agent (times concurrency), one task per timer, and starts the shared storage manager.
3. Each consumer joins its consumer group with XREADGROUP, wraps every entry in a Message (topic, stream id, decoded fields, send/receive timestamps) and awaits your handler.
4. SIGINT/SIGTERM trigger a graceful shutdown: timers stop, tasks are cancelled with a timeout, Redis connections and the storage manager are closed.

Run several copies of the same script to scale horizontally; Redis distributes entries across consumers in a group.

Documentation

• Getting started
• Configuration
• Architecture
• Scaling
• Best practices
• Testing
• Examples
• LLM_API.md: condensed API reference intended for pasting into an LLM context

Development

git clone https://github.com/trbck/streammachine.git
cd streammachine
python -m venv .venv && source .venv/bin/activate
pip install -e ".[dev,all]"
pytest                                  # unit tests (no Redis needed)
RUN_INTEGRATION_TESTS=1 pytest          # integration tests against a local Redis
ruff check src tests

Build and check the distribution:

python -m build
twine check dist/*

License

Apache License 2.0. See LICENSE.