Hi Team - I'd like to add this function in a utility script

from prefect.tasks.notifications import SlackTask
class Alert:

    def alert_on_special_failure(task, old_state, new_state):
        if new_state.is_failed():
            if getattr(new_state.result, "flag", False) is True:
                errMsg = '--- LOADER ERROR ---'
                msg = "{}\nTask: `{}` FAILED.\nThe loader process failed: `{}`".format(
                        errMsg, task.name, new_state.result.value)
                SlackTask().run(message = msg, webhook_secret = "SLACK_PREFECT_DEV")
        return new_state

This was previous called like this:

@task(name = 'send API request',
      max_retries = 3,
      retry_delay = timedelta(minutes = 5),
      state_handlers = [alert_on_special_failure])
def post_request_process(

If we add this alert on special failure object, will we need to pass in the task, old_state, and new_state? Were these variables global?

Sweet, glad you got it figured out Chris. If you are switching over to 2.0 definitely check out the slack block and new slack collections integration. https://prefecthq.github.io/prefect-slack/ https://docs.prefect.io/api-ref/prefect/blocks/notifications/