Basic flow implementation

For both run methods, you'll have to pass in env as a parameter or use self.env in the code.