Fabien Punin
03/23/2026, 9:02 PMMarvin
03/23/2026, 9:02 PMMarvin
03/23/2026, 9:07 PMflow, flow_run, and state. But you can still access deployment context from inside the hook.
Ways to get deployment info in an on_completion hook (Prefect 3.x):
- From the runtime context:
- prefect.runtime.deployment exposes id, name, version, parameters (None/empty if not running from a deployment)
- prefect.runtime.flow_run exposes run fields like id, name, parameters, etc.
- Docs: Runtime context, runtime.deployment API, runtime.flow_run API
- From the flow_run argument:
- flow_run.deployment_id, flow_run.deployment_version, plus other metadata fields
- If you need to pass extra, custom data to the hook:
- Pre-bind it when registering the hook (via functools.partial)
- Or stash it during the flow run and read it in the hook (same process only)
References for hooks: State-change hooks, Flow decorator and hooks API
Examples
- Access deployment context directly in the hook
from prefect import flow
from prefect.runtime import deployment
def on_complete(flow, flow_run, state):
# Via runtime context (works only if triggered from a deployment)
print("Deployment name:", deployment.name)
print("Deployment id:", deployment.id)
# Via flow_run object
print("Deployment id from flow_run:", flow_run.deployment_id)
print("Deployment version from flow_run:", flow_run.deployment_version)
@flow(on_completion=[on_complete])
def my_flow():
...
- Pass extra configuration to the hook
from functools import partial
from prefect import flow
def notifier(flow, flow_run, state, team="data-eng"):
print(f"[{team}] Flow {flow.name} finished in state {state.name}")
@flow(on_completion=[partial(notifier, team="mlops")])
def my_flow():
...
- Stash data during the run for the hook to read later (same process)
from prefect import flow
from prefect.context import get_run_context
def on_complete(flow, flow_run, state):
cfg = flow_run.context.get("hook_cfg", {})
print("Hook cfg:", cfg)
@flow(on_completion=[on_complete])
def my_flow():
ctx = get_run_context()
ctx.flow_run.context["hook_cfg"] = {"alert_level": "critical"}
...
If you can share which “deployment context” fields you need (e.g., name, id, parameters, work pool/queue), I can suggest the most direct access pattern.Fabien Punin
03/23/2026, 9:14 PMMarvin
03/23/2026, 9:14 PMMarvin
03/23/2026, 9:14 PMflow, flow_run, state), you can get the deployment name inside the hook.
Fast path (when running from a deployment):
from prefect.runtime import deployment
def on_complete(flow, flow_run, state):
dep_name = deployment.name # None if not running via a deployment
print("Deployment name:", dep_name)
Robust path (fallback to the API if needed):
from prefect.runtime import deployment
async def on_complete(flow, flow_run, state):
name = deployment.name
if not name and flow_run.deployment_id:
from prefect.client.orchestration import get_client
async with get_client() as client:
dep = await client.read_deployment(flow_run.deployment_id)
name = dep.name
print("Deployment name:", name)
Docs:
- State-change hooks
- Runtime context
- runtime.deployment API