Great question. In Prefect 3.x you can do two things: 1) Attach a flow “on cancel” hook that always runs when the flow enters the Cancelling state (recommended) 2) Optionally, add your own custom event emission when that happens Here’s a minimal example that does both. - Registers a cancellation hook - Emits a custom event when cancellation starts - Leaves a place to do cleanup (close connections, delete temp files, etc.)

from prefect import flow
from prefect.events.utilities import emit_event

def on_flow_cancel(flow, flow_run, state):
    # This is called as soon as the flow enters 'Cancelling'
    # Keep this fast and resilient
    print(f"Flow {flow_run.name} is being cancelled (id={flow_run.id})")

    # Emit a custom event you can subscribe to or use in Automations
    emit_event(
        event="example.flow.cancelled",
        resource={
            "prefect.resource.id": f"prefect.flow-run.{flow_run.id}",
            "prefect.resource.role": "flow-run",
        },
        payload={"flow_run_name": flow_run.name, "state_message": state.message},
    )

    # Do any cleanup you need here (notify, close resources, etc.)
    # e.g., close_db_connection(), cleanup_temp_files(), etc.

@flow(on_cancellation=[on_flow_cancel])
def my_flow():
    # Your normal flow logic
    import time
    print("Starting long-running work...")
    time.sleep(600)  # Simulate long work

Notes: - The

on_cancellation

hook runs when the flow transitions to Cancelling, before it fully terminates, giving you a reliable place to run cleanup and/or emit a custom event. - Prefect already emits standard flow-run state events you can target with Automations. The custom

emit_event

above is optional, useful if you want your own event name/payload. If you prefer inline cleanup in your own code (on top of the hook), use try/finally to guarantee local cleanup:

from prefect import flow

@flow
def my_flow():
    resource = acquire_something()
    try:
        do_work(resource)
    finally:
        # This runs even if the flow is cancelled
        resource.close()

How to cancel a running flow - From the UI: open the flow run and click Cancel - From the CLI (verified):

prefect flow-run cancel <FLOW_RUN_ID>

Docs to dig deeper: - State-change hooks: Flow and task state-change hooks - Cancelling workflows: Cancel workflows - Emitting events: prefect.events.utilities.emit_event If you can share whether you want to take action inside the flow code vs. trigger an external system (e.g., via an Automation), I can tailor the example further.

Do you mean logging inside a Prefect flow’s on_cancellation hook? If so, here’s the pattern in Prefect 3.x. Key points: - Only flows have an

on_cancellation

hook; tasks do not. - Use

flow_run_logger(flow_run)

inside the hook to ensure logs are attributed to the run and appear in the UI. - Cancellation/crashed hooks are controlled by

PREFECT__ENABLE_CANCELLATION_AND_CRASHED_HOOKS

(defaults to true). Example (sync):

from prefect import flow
from prefect.logging.loggers import flow_run_logger

def on_cancel(flow_obj, flow_run, state):
    logger = flow_run_logger(flow_run)
    logger.warning(f"Flow {flow_obj.name} is being cancelled. State: {state.type}")

@flow(on_cancellation=on_cancel)
def my_flow():
    # your flow logic
    ...

if __name__ == "__main__":
    my_flow()

Example (async hook works too):

from prefect import flow
from prefect.logging.loggers import flow_run_logger

async def on_cancel(flow_obj, flow_run, state):
    logger = flow_run_logger(flow_run)
    <http://logger.info|logger.info>("Cleanup or notifications here...")

@flow(on_cancellation=on_cancel)
def my_flow():
    ...

Environment variable (usually not needed, it’s true by default):

export PREFECT__ENABLE_CANCELLATION_AND_CRASHED_HOOKS=true

If you wanted logging during task/flow cleanup on cancellation, a

finally

block with

get_run_logger()

also works inside the task/flow body:

from prefect import task, get_run_logger

@task
def do_work():
    logger = get_run_logger()
    try:
        ...
    finally:
        <http://logger.info|logger.info>("Cleaning up after cancellation or error")

Docs: - State change hooks - Add logging - Logging APIs If you meant a different “cancellation function” (e.g., a task-level hook or a non-Prefect callback), let me know the context and I’ll tailor the example.

In a flow’s on_cancellation hook, you can read the parameters directly from the

flow_run

object the hook receives. Example:

from prefect import flow
from prefect.logging.loggers import flow_run_logger

def on_cancel(flow_obj, flow_run, state):
    logger = flow_run_logger(flow_run)
    params = flow_run.parameters or {}
    logger.warning(f"Flow {flow_obj.name} is being cancelled with params: {params}")

@flow(on_cancellation=on_cancel)
def my_flow(a: int, b: str = "default"):
    ...

Other options: - From anywhere inside the running flow body:

from prefect.runtime import flow_run
  # dict of parameters for the current flow run
  params = flow_run.parameters

- If you prefer context access inside the flow body:

from prefect.context import get_run_context
  params = get_run_context().parameters

Notes: - Only flows have

on_cancellation

hooks; tasks do not. - Cancellation/crashed hooks are controlled by

PREFECT__ENABLE_CANCELLATION_AND_CRASHED_HOOKS

(defaults to true). - Be careful not to log sensitive values if your params include secrets. Docs: - State change hooks