Your workflows. Your infrastructure. Your data.

@workflow(name="order-processing")
class OrderWorkflow:
    async def run(self, ctx: WorkflowContext, input: OrderInput) -> OrderOutput:
        # Schedule payment task (string-based task reference)
        payment = await ctx.execute_task("payment-task", {
            "amount": input.amount,
            "currency": input.currency
        })

        # Schedule inventory task
        inventory = await ctx.execute_task("inventory-task", {"items": input.items})
        if inventory["out_of_stock"]:
            await ctx.run("refund", lambda: refund_payment(payment["transaction_id"]))
            raise NonRetryableError("Inventory unavailable")

        return OrderOutput(order_id=input.order_id, status="completed")