Jørgen Nygaard
11/21/2025, 11:18 AMMarvin
11/21/2025, 11:18 AMMarvin
11/21/2025, 11:22 AM(flow, flow_run, state). You can’t pass arbitrary, ad-hoc args into a hook at call time via the decorator. The usual way to “parameterize” a hook is either:
- Bind kwargs with functools.partial when attaching the hook (e.g., via with_options), or
- Read whatever you need from runtime objects available to the hook.
Good news: you can get most runtime data inside the hook without `with_options`:
- `flow_run.parameters`: all parameters passed to the flow run
- `flow_run.tags`: tags on the run
- `flow_run.name`: run name
- `get_run_context()`: full runtime context; includes ctx.parameters, ctx.flow_run, ctx.flow, and a client you can use to fetch Variables, etc.
Examples
- Attach via decorator and read runtime data in the hook:
from prefect import flow
from prefect.context import get_run_context
def notify_on_failure(flow, flow_run, state):
# Access runtime data without passing custom args
params = flow_run.parameters # or get_run_context().parameters
tags = flow_run.tags
run_name = flow_run.name
# do something with params/tags/run_name/state.message ...
@flow(on_failure=[notify_on_failure])
def my_flow(user_id: str, priority: str = "normal"):
raise RuntimeError("boom")
my_flow(user_id="abc123", priority="high")
- If you need truly custom per-run data that isn’t a flow param, you can:
- Put it in a tag and read flow_run.tags
- Set it as part of the run’s context when creating the run via API (if you’re starting runs programmatically), then read get_run_context().flow_run.context
- Store it in a Prefect Variable or env var and read it inside the hook
What you cannot do
- You can’t pass extra positional/keyword args to a hook attached via @flow(..., on_failure=[...]) or @my_flow.on_failure at run time.
- You can’t register hooks dynamically after execution starts; they must be attached before the state change occurs.
Docs
- State change hooks guide: https://docs.prefect.io/latest/how-to/workflows/state-change-hooks/
- Runtime context (get_run_context): https://docs.prefect.io/latest/api-ref/context/#prefect.context.get_run_context
If you can share what custom data you’re trying to pass, I can suggest the cleanest pattern (e.g., parameters vs tags vs variables vs context) for your setup.Jørgen Nygaard
11/21/2025, 11:26 AMMarvin
11/21/2025, 11:26 AMMarvin
11/21/2025, 11:29 AMflow_run.context.
- Use flow parameters: Pass any per-run data as parameters; your hook can read them from flow_run.parameters or get_run_context().parameters.
- Use job variables: Deployment-level values are available via runtime and can be read in hooks.
- Persist mutable data externally: If you need mutable, ad-hoc data that changes during the run and is later read by a hook, write it to an external store keyed by flow_run.id (e.g., DB/Redis/object storage) and read it in the hook.
Example hook accessing runtime data without custom args:
from prefect import flow
from prefect.context import get_run_context
def on_fail(flow, flow_run, state):
ctx = get_run_context()
# Read-only views of runtime data
params = ctx.parameters # same as flow_run.parameters
tags = flow_run.tags
creation_context = flow_run.context # set at flow-run creation time
# fetch mutable extras from external store if needed using flow_run.id
# ...
@flow(on_failure=[on_fail])
def my_flow(user_id: str, feature_flag: str):
raise RuntimeError("boom")
If you’re starting runs via the API and want to seed flow_run.context, you can pass a dict at creation time; your hook can read it later. But there’s no supported API to update flow_run.context mid-run. If you tell me what data you want to stash and when you need it, I can suggest the cleanest pattern (params vs context-at-creation vs external store).