Python

Python is a first-class durable runtime, not just a place to run a step. The durable-worker SDK (the same one the aviary/nestjs-durable ecosystem ships) speaks the wire your Adonis engine speaks byte-for-byte, so one Python worker can do either — or both:

• Implement steps an AdonisJS workflow calls — a ctx.step('payments.charge-card', …) reaches a handler written in Python.
• Author whole workflows in Python — the coordinator logic (ctx.step, ctx.sleep, ctx.gather_calls, …) lives in Python, and the Adonis engine drives it turn by turn.

Either way the engine stays the sole owner of durable state, recovery, and timers — the Python worker is a stateless consumer on the broker. Everything rides the same Redis queues the BullMQ transport uses.

pip install "durable-worker[redis]"

A Python step worker

import asyncio
from durable_worker import Worker, FatalError
from durable_worker.redis_runner import run_redis_worker

worker = Worker()

@worker.step("payments.charge-card")
async def charge(data):
    res = await stripe.charge(data["orderId"], data["amountCents"])
    return {"chargeId": res.id}

async def main():
    await run_redis_worker(worker, connection="redis://localhost:6379")
    await asyncio.Event().wait()

asyncio.run(main())

The handler's argument is the step input (schema-validated by the engine); its return value is the output. Raise FatalError for a non-retryable failure (e.g. a declined card); any other exception is retryable and the engine applies the step's retry policy.

Connecting to the control plane

There is no direct connection between the Python worker and your AdonisJS app — everything goes through the broker (Redis). Point the worker at the same Redis your engine's bullmq transport uses, and register a handler under the exact same name. Two things have to line up:

import { defineConfig, transports, stores } from '@adonis-agora/durable'

export default defineConfig({
  role: 'control-plane',
  transport: 'bullmq',
  transports: { bullmq: transports.bullmq({ connection: { host: '127.0.0.1', port: 6379 } }) },
  store: 'lucid',
  stores: { lucid: stores.lucid({ connection: 'pg' }) },
})

Each registered step name gets its own queue — no separate group to declare. Sharing one broker across isolated deployments? Set the same partition on both sides (transports.bullmq({ partition }) engine-side, run_redis_worker(worker, partition=...) worker-side) and every queue is suffixed <name>@<partition>.

Authoring workflows in Python

A Python worker can own the whole workflow function, not just a step. The Adonis engine stays the sole owner of durable state, recovery, and timers: it advances a run one turn at a time by sending the run's history as a workflow task, the Python runtime replays the workflow function locally, and returns the commands the replay produced (a step dispatch, a sleep, a signal wait) for the engine to persist and act on — a pure, store-free coordinator.

The simplest deployment registers the workflow and its step handlers on one Worker (each registered name gets its own queue — route-by-handler), run with run_workers:

from durable_worker import Worker, WorkflowContext, redis_url_from_env, run_workers

worker = Worker()

@worker.step("ingestion")
async def ingestion(data):
    return {"rows": load(data["key"])}

@worker.workflow("pipeline")
def pipeline(ctx: WorkflowContext, data):
    started_at = ctx.now()                               # replay-stable capture
    results = ctx.gather_calls([                          # fan out parallel remote steps in this run
        {"name": "ingestion", "input": {"key": f"/{data['id']}/a.csv"}},
        {"name": "ingestion", "input": {"key": f"/{data['id']}/b.csv"}},
    ])
    ctx.sleep(60_000)                                    # durable timer
    return {"rows": sum(r["rows"] for r in results), "startedAt": started_at}

run_workers([worker], redis=redis_url_from_env())        # one loop, one Redis pool, graceful shutdown

ctx mirrors the TypeScript WorkflowCtx — each op is keyed by a deterministic seq, so a completed op replays its recorded result instead of re-running:

Changing the op sequence under a run already in flight raises NondeterminismError. A failed step raises StepFailed; the gather* fan-outs default to wait_all (raising GatherFailed), with an opt-in fail_fast.

Dispatching a Python workflow from Adonis

Start the workflow by its name — whichever live Python worker is serving that name picks up the turns. From an AdonisJS service:

import engine from '@adonis-agora/durable/services/main'

await engine.start('pipeline', { id: baseId }, `pipeline-${baseId}`)

or, from inside another workflow, await ctx.child('pipeline', { id: baseId }). The queue name is the routing: the worker serving pipeline runs it. If none is live, the start fails fast so a typo'd name never silently hangs.

The workflow-task/decision wire is language-neutral, so the same Python workflow runs unchanged against an Adonis, a NestJS, or any other conformant control plane.