In Prefect 2.x, you attach an on_failure hook as a callable. The hook receives three arguments from Prefect automatically: - For flows:

(flow, flow_run, state)

- For tasks:

(task, task_run, state)

To pass your own arguments, pre-bind them with functools.partial (or use defaults/kwargs in your hook). Example: flow-level hook with a custom channel

from functools import partial
from prefect import flow

def notify_slack(flow, flow_run, state, channel="#alerts"):
    print(f"[{channel}] Flow {flow_run.name} failed with state={state.name} msg={state.message}")

notify_ops = partial(notify_slack, channel="#data-alerts")

@flow(on_failure=[notify_ops])
def my_flow():
    1 / 0

Example: task-level hook with a custom owner

from functools import partial
from prefect import task

def on_task_fail(task, task_run, state, owner=None):
    print(f"Owner={owner} | Task {task_run.name} failed with state={state.name} msg={state.message}")

notify_owner = partial(on_task_fail, owner="alice")

@task(on_failure=[notify_owner])
def risky():
    1 / 0

Per-call customization with with_options (useful when you can’t set it at decoration time):

risky_for_bob = risky.with_options(on_failure=[partial(on_task_fail, owner="bob")])

Notes - You can define your hook as

def hook(..., **kwargs)

and pass any extra keywords via

partial(...)

. - If you need dynamic data at runtime, read it from the provided objects instead of passing it in: - Flow: use

flow_run.id

,

flow_run.name

,

flow_run.tags

, possibly

flow_run.parameters

- Task: use

task_run.id

,

task_run.name

- State: use

state.name

,

state.message

Docs: - State change hooks guide - States concepts If you can share whether you’re adding the hook to a flow or a task, and what argument you want to pass (static vs. computed at runtime), I can tailor the example further.

Great question. In Prefect 2.x, you don’t “pass” a flow argument into a hook at registration time (since parameters are only known at runtime). Instead, read the parameter from the flow run that Prefect passes to your hook. Flow-level hook: read from flow_run.parameters

from prefect import flow

def notify_on_failure(flow, flow_run, state):
    env = flow_run.parameters.get("env")  # this is your flow argument
    print(f"[{env}] Flow {flow_run.name} failed: {state.message}")

@flow(on_failure=[notify_on_failure])
def my_flow(env: str):
    1 / 0  # force a failure

Task-level hook (when you need the flow’s parameter inside a task hook): use get_run_context().parameters

from prefect import task, get_run_context

def on_task_failure(task, task_run, state):
    ctx = get_run_context()
    env = ctx.parameters.get("env")
    print(f"[{env}] Task {task_run.name} failed: {state.message}")

@task(on_failure=[on_task_failure])
def risky():
    1 / 0

@flow
def parent(env: str):
    risky()

Key points - Don’t use partial for runtime flow parameters; use the provided

flow_run

(flow hooks) or

get_run_context().parameters

(task hooks). - Parameters are resolved values for the run, so you’ll get exactly what the flow was called with. Docs: - State change hooks guide - States concepts