Consume Redis Streams without writing the loop.

RillKit is a Python library on top of redis-py. You name a stream and a group, decorate a function and call run(). Consumer groups, retries with backoff, a dead-letter queue and a clean shutdown come with it.

Get started Source

pip install rillkit

Python 3.9+ MIT one dependency: redis-py

python worker.py

12:04:01INFOconsumer started | stream: orders | group: workers | policy: nack_on_error | max retries: 3
12:04:02stdoutcharged order 1
12:04:02stdoutcharged order 2
12:04:03ERRORhandler failed for message 1730032643012-0 (delivery 1): card declined
12:04:09ERRORhandler failed for message 1730032643012-0 (delivery 2): card declined
12:04:21stdoutcharged order 3
12:04:21WARNINGmoving message 1730032645250-0 to orders-dlq: bad total
12:04:22

Two files and a stream name

The publisher and the worker agree on "orders" and nothing else. No schema to register, no polling loop, no ack bookkeeping. The path below is the one message 42.5 takes.

publish.pypython

from rillkit import Broker

broker = Broker.from_env()

broker.publish("orders", {"id": 1, "total": 42.5})

worker.pypython

from rillkit import AckPolicy, Broker, Message, Stream

broker = Broker.from_env()

@broker.subscribe(
    Stream("orders", group="workers"),
    ack_policy=AckPolicy.NACK_ON_ERROR,
    max_retries=3,
)
def handle(order: dict, msg: Message):
    if order["total"] <= 0:
        msg.reject("bad total")  # straight to the DLQ
        return
    if not charge(order):
        msg.nack()  # try again after a backoff

broker.run()

Where a message ends up

It goes from publish to the handler, then takes one of three exits. There is no fourth one where it quietly disappears.

AckThe handler returned. The entry is acked and leaves the pending list.
RetryThe handler raised, or called nack(). It is delivered again after a growing backoff, up to max_retries.
Dead letterRetries ran out, or you called reject(). The entry moves to the DLQ with the error, original id and delivery count.

Nothing hidden. Look in Redis.

Groups, pending entries and the dead-letter stream are ordinary Redis structures. When something looks off at 2am, redis-cli shows you what RillKit sees.

Acked: The entry is acked with XACK and leaves the pending list. lag: 0 means the group has read everything.
Waiting to retry: Still pending, with a delivery count. It is claimed again once it has been idle for claim_idle_ms times the backoff.
Dead-lettered: Copied to orders-dlq with the original body plus five x- fields saying where it came from and why it stopped.

redis-cli

127.0.0.1:6379> XINFO GROUPS orders
1)  1) "name"
    2) "workers"
    3) "consumers"
    4) (integer) 2
    5) "pending"
    6) (integer) 1
    7) "last-delivered-id"
    8) "1730032645250-0"
    9) "entries-read"
   10) (integer) 7
   11) "lag"
   12) (integer) 0

127.0.0.1:6379> XPENDING orders workers - + 10
1) 1) "1730032643012-0"
   2) "workers-web1-4242"
   3) (integer) 6210
   4) (integer) 2

127.0.0.1:6379> XRANGE orders-dlq - + COUNT 1
1) 1) "1730032645261-0"
   2) 1) "data"
      2) "{\"id\": 7, \"total\": -5}"
      3) "x-original-id"
      4) "1730032645250-0"
      5) "x-stream"
      6) "orders"
      7) "x-group"
      8) "workers"
      9) "x-error"
     10) "bad total"
     11) "x-deliveries"
     12) "1"

The loop you would have written

Most teams start with xreadgroup in a while True. It works until the first exception. The migration example runs both versions side by side.

consumer_raw.pypython

import json

import redis

r = redis.Redis.from_url("redis://localhost:6379/0", decode_responses=True)
try:
    r.xgroup_create("orders", "workers", id="0", mkstream=True)
except redis.ResponseError as e:
    if "BUSYGROUP" not in str(e):
        raise

while True:
    for _, entries in r.xreadgroup("workers", "web1", {"orders": ">"}, count=10, block=5000) or []:
        for entry_id, fields in entries:
            try:
                charge(json.loads(fields["data"]))
                r.xack("orders", "workers", entry_id)
            except Exception as e:
                print("failed:", e)  # and then?

worker.pypython

from rillkit import Broker, Stream

broker = Broker.from_env()

@broker.subscribe(Stream("orders", group="workers"))
def handle(order: dict):
    charge(order)

broker.run()

Entries stuck with a crashed worker are never claimed again.
A handler that keeps failing keeps failing. There is no retry limit and no backoff.
Nothing goes to a dead-letter stream, so bad messages pile up in the pending list.
Ctrl-C drops whatever was in flight.

What happens when a handler fails

Consumer groups
The group is created for you and every worker gets its own consumer name. Entries left with a crashed worker are claimed back once they have sat idle long enough.
Retries with backoff
A failing message comes back after an exponential backoff, up to max_retries. The default is 3, so a handler runs at most four times for one message.
nack or reject
msg.nack() asks for another attempt. msg.reject("reason") says this one can never work and skips the remaining retries.
Dead-letter stream
Entries that run out of retries, or are rejected, move to <stream>-dlq with the error, original id, group and delivery count. broker.replay_dlq() puts them back.

Ack policies

Choose when a message counts as done. Set it per handler with ack_policy=. Details

Policy	Handler returns	Handler raises
`NACK_ON_ERROR`default	ack	retry	At-least-once. Failures are retried, then dead-lettered.
`ACK_ALWAYS`	ack	ack	Failures are logged and dropped, never retried.
`ACK_FIRST`	acked before	acked before	At-most-once. A crash mid-handler loses the message.
`MANUAL`	you call msg.ack()	retry	RillKit never acks. Return without acking and it is retried.

Around the handler

Middleware: Hooks for receiving, handling, publishing and cleanup. Idempotent ships in the box and skips duplicate deliveries.
Startup and shutdown hooks: on_startup, after_startup, on_shutdown and after_shutdown, as plain decorators on the broker.
Dicts or Pydantic: Handlers get the decoded JSON as a dict. Annotate the parameter with a Pydantic model and it is validated before your code runs. Pydantic is optional.
Graceful shutdown, heartbeat: SIGINT and SIGTERM let in-flight messages finish. Each consumer refreshes a heartbeat key, and active_workers() reads them.
Plain wire format: One JSON string in the data field of an ordinary stream entry. Anything that can XADD can publish, and anything that can XREADGROUP can read.

Running in about five minutes

You need Python 3.9 or newer and a Redis you can reach. Local Docker is fine. The guide walks through the same steps with more explanation, then covers retries, the DLQ and running several workers.

Read the getting-started guide Examples API reference

terminal

# redis, if you do not have one
$ docker run -d -p 6379:6379 redis:7

$ pip install rillkit
$ export REDIS_URL=redis://localhost:6379/0

$ python worker.py   # terminal 1
$ python publish.py  # terminal 2

Consume Redis Streams without writing the loop.

Nothing hidden. Look in Redis.

The loop you would have written

What happens when a handler fails

Consumer groups

Retries with backoff

nack or reject

Dead-letter stream

Ack policies

Around the handler

Running in about five minutes