parakeet-workflows 0.1.0

🦜🔀 Parakeet Workflows

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

✨ Highlight features

• Event-driven execution — Steps communicate through typed events.
• Fan-out / Fan-in — Emit multiple events in parallel and join results back, with full async support.
• Shared state — Pass data across steps via a built-in Context object without global variables.
• Internal buffer — Events are queued internally, so steps can produce and consume at their own pace.
• Declarative API — Define steps with a simple @step decorator.
• Built-in observability — Instrument workflows with OpenTelemetry-compatible tracing and custom metrics.

📦 Installation

pip install parakeet-workflows

With uv:

uv add parakeet-workflows

🚀 Quickstart

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

import asyncio
from parakeet_workflows import Workflow, Context, step
from parakeet_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

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

import asyncio
from parakeet_workflows.server import WorkflowServer

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

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

License

Apache License 2.0