How to write a Custom Action

  1. Write a class inherited from mistral_lib.actions.Action

from mistral_lib import actions

class RunnerAction(actions.Action):
    def __init__(self, param):
        # store the incoming params
        self.param = param

    def run(self, action_ctx=None):
        # return your results here
        return {'status': 0}

  1. Publish the class in a namespace (in your setup.cfg)

[entry_points]
mistral.actions =
    example.runner = my.mistral_plugins.somefile:RunnerAction

  1. Install the Python package containing the action. If this was added to Mistral itself it will need to be reinstalled.

  2. Restart Mistral services (API, engine and executor) so they load the new action. Actions plugged in with the mistral.actions entry point are discovered when a service starts, there is no need to populate the database.

  3. Now you can call the action example.runner

my_workflow:
  tasks:
    my_action_task:
      action: example.runner
      input:
        param: avalue_to_pass_in