Overview

High performance and flexible event-driven workflow engine, designed to build complex tasks.

---

Installation

pipuv

$ pip install donkey-workflows

$ uv add donkey-workflows

Quick Start

Here's a simple example to get you started with Donkey Workflows:

import asyncio
from donkey_workflows import Workflow, Context, step
from donkey_workflows.events import Event, StartEvent, StopEvent


class MessageEvent(Event):
    message: str

class MyWorkflow(Workflow):

    @step(when=StartEvent)
    async def start(self, ctx: Context, ev: StartEvent) -> MessageEvent:
        input_msg = ev.get("message", "")
        return MessageEvent(message=f"Processed: {input_msg}")

    @step(when=MessageEvent)
    async def process(self, ctx: Context, ev: MessageEvent) -> StopEvent:
        return StopEvent(result=ev.message)


async def main():
    workflow = MyWorkflow()
    result = await workflow.run(input_msg="Hello, World!")
    print(result)

asyncio.run(main())

Core Concepts

Workflow

A workflow is a class that inherits from Workflow and contains one or more steps. It orchestrates the execution of steps based on events.

Steps

Steps are asynchronous methods decorated with @step(when=EventType) that define what happens when a specific event is received.

• Steps receive a Context and an Event.
• Steps can return new events to trigger subsequent steps.

Events

Events are the building blocks of workflows. They carry data between steps and trigger step execution.

• StartEvent: Automatically triggered when a workflow starts
• StopEvent: Signals the end of a workflow and carries the final result
• Custom Events: Define your own events by inheriting from Event

Context

The Context object provides access to workflow state and allows steps to share data throughout the workflow execution.

# Read-only access
current_value = ctx.state.count

# Edit state
async with ctx.store.edit_state() as state:
    state.count = current_counter + 1

# Send events
ctx.send_event(MyEvent(...))

Server

Donkey Workflows includes an optional HTTP server built on FastAPI that exposes your workflows as REST endpoints.

import asyncio
from donkey_workflows.server import WorkflowServer

server = WorkflowServer()
server.add_workflow("my-workflow", MyWorkflow())

asyncio.run(server.serve(host="0.0.0.0", port=8080))

Endpoint	Description
GET /workflows	List all registered workflows
POST /workflows/{id}/run	Execute a workflow

Features

Feature	Donkey Workflows
Event-driven execution	✅
Fan-out (parallelism)	✅
Fan-in (joining)	✅
Async execution	✅
Shared state	✅
Internal buffer	✅
Declarative API	✅
Observability	✅

License

Apache License 2.0