Currently I have this:

def escalate(obj, old_state, new_state):
    if new_state.is_failed() and new_state.message not in IGNORE_MESSAGES:
        msg = f"{obj} {new_state.message}"
        slack_post(msg, channel="#errors", username="Prefect")
    return new_state

and i would like to

def escalate(obj, old_state, new_state, error_channel="#errors"):

because i need have different channel for some flows

check this post for all details about parametrization in 1.0 https://www.prefect.io/guide/blog/how-to-make-your-data-pipelines-more-dynamic-using-parameters-in-prefect/

I believe this is the code snippet you are looking for from the article Anna posted:

import prefect
from prefect import Flow, task, Parameter
from prefect.tasks.prefect import RenameFlowRun


def rename_handler(obj, new_state, old_state):
    if new_state.is_running():
        user = prefect.context.parameters.get("user_name")
        param_value = user.lower().replace(" ", "_")
        RenameFlowRun().run(flow_run_name=f"hello_{param_value}")
    return


@task(log_stdout=True)
def greet_user(user_name):
    print(f"Hello {user_name}")


with Flow("using_parameters_in_state_handler", state_handlers=[rename_handler]) as flow:
    param = Parameter("user_name", default="Jerry Seinfeld")
    greet_user(param)

Basically, you can pass a parameter to your state handler using

Parameter

Thanks @Anna Geller and @Khuyen Tran. But i don't want to pass the parameter to each function. Suppose i have a prefect flow script which contains 6 tasks and each task one state handler. I need to pass name to state handler by not passing the parameter to each function

What problem are you trying to solve?

def escalate(obj, old_state, new_state):
    if new_state.is_failed() and new_state.message not in IGNORE_MESSAGES:
        msg = f"{obj} {new_state.message}"
        slack_post(msg, channel="#errors", username="Prefect")
    return new_state

i need to access param = Parameter("slack_channel_name", default="#api-errors") in the escalate function

Ahh gotcha, try using prefect.context.parameters.get("slack_channel_name")

where?. In below?

def main():
    flow = Flow("Test data ingestion", schedule=schedule, state_handlers=[escalate])
    param = Parameter("slack_channel_name", default="#api-errors")
    flow.add_task(test_script1)
    flow.set_dependencies(
        test_script2, upstream_tasks=[test_script1]
    )

    register_flow(
        flow=flow,
        project_name=PIPELINE_PROJECT,
    )

In your state handler function

thanks