Simplest pattern in Prefect 2.20: use a flow-level on_completion hook and update the run via the orchestration client. Key notes: -

update_flow_run(..., tags=...)

replaces all existing tags; merge first if you want to preserve them. - Hook signature is

(flow, flow_run, state)

and can be async. Example (adds a “completed” tag while preserving any existing tags):

from prefect import flow
from prefect.client.orchestration import get_client

async def tag_on_completion(flow, flow_run, state):
    # preserve existing tags, add new ones
    new_tags = sorted(set(flow_run.tags) | {"completed"})
    async with get_client() as client:
        await client.update_flow_run(flow_run.id, tags=new_tags)

@flow(on_completion=[tag_on_completion])
def my_flow():
    # your work here
    return "done"

If you want different tags for success vs failure, use both

on_completion

and

on_failure

, or check

state.is_completed()

/

state.is_failed()

inside the hook. Want me to adapt this to add tags conditionally (e.g., based on parameters or runtime info)?

Great question. In Prefect 2.20 you’ve got two practical options to tag a flow run based on the outputs of tasks: Option A (simplest): Compute tags inside the flow using the task results you already have, then update the flow run’s tags before the flow completes. - No hooks needed - No round-trips to the API - Works even if you don’t persist task results Example:

from prefect import flow, task, get_run_context
from prefect.client.orchestration import get_client

@task
def compute_score(x: int) -> int:
    return x * 2

@task
def decide_label(score: int) -> str:
    return "high" if score >= 10 else "low"

@flow
def my_flow(x: int = 5):
    score_fut = compute_score(x)
    label_fut = decide_label(score_fut)

    # materialize task outputs within the flow
    score = score_fut.result()
    label = label_fut.result()

    # derive tags
    new_tags = {f"score-{score}", f"label-{label}"}

    # merge with any existing flow run tags
    ctx = get_run_context()
    merged = sorted(set(ctx.flow_run.tags or []) | new_tags)

    # update tags synchronously
    with get_client(sync_client=True) as client:
        client.update_flow_run(ctx.flow_run.id, tags=merged)

    return {"score": score, "label": label}

if __name__ == "__main__":
    my_flow()

Option B (using a hook): Read task runs (and optionally their results) in an on_completion hook and set tags. - Useful if you prefer post-run tagging or want to base tags on overall execution state - If you need actual task outputs (not just success/failure), ensure those tasks persist results (e.g.,

@task(persist_result=True)

), otherwise the hook may not be able to materialize them Example:

from prefect import flow, task
from prefect.client.orchestration import get_client
from prefect.client.schemas.filters import FlowRunFilter, FlowRunFilterId

@task(persist_result=True)
def compute_score(x: int) -> int:
    return x * 2

@task(persist_result=True)
def decide_label(score: int) -> str:
    return "high" if score >= 10 else "low"

async def tag_from_task_outputs(flow, flow_run, state):
    async with get_client() as client:
        # fetch all task runs for this flow run
        task_runs = await client.read_task_runs(
            flow_run_filter=FlowRunFilter(id=FlowRunFilterId(any_=[flow_run.id]))
        )

        derived = set()

        for tr in task_runs:
            # get final state to attempt to read the result
            states = await client.read_task_run_states(task_run_id=tr.id)
            final = states[-1] if states else None

            # fall back to status-only tags if no result
            if not final:
                derived.add(f"{tr.task_key}-{tr.state_type or 'UNKNOWN'}")
                continue

            if getattr(final, "is_completed", lambda: False)():
                try:
                    # will only work if results were persisted
                    value = final.result(raise_on_failure=False, fetch=True)
                    # make a short, safe tag from the value
                    derived.add(f"{tr.task_key}={str(value)[:20]}")
                except Exception:
                    # if result isn't available, tag by state
                    derived.add(f"{tr.task_key}-COMPLETED")
            else:
                derived.add(f"{tr.task_key}-{final.type.value}")

        # merge with existing tags on the flow run
        merged = sorted(set(flow_run.tags or []) | derived)

        # IMPORTANT: update_flow_run(tags=...) replaces all tags
        await client.update_flow_run(flow_run.id, tags=merged)

@flow(on_completion=[tag_from_task_outputs])
def my_flow(x: int = 5):
    s = compute_score(x)
    decide_label(s)

Notes: -

update_flow_run(tags=...)

replaces all tags; merge with existing tags if you want to preserve them (as shown). - In hooks, prefer

async with get_client() as client: ...

.

- To base tags on actual task return values in a hook, ensure those tasks persist results (e.g.,

@task(persist_result=True)

), or configure results persistence globally. Otherwise you can still tag based on task states (COMPLETED/FAILED). If you can share the exact tagging logic (e.g., which task outputs and what tags you want), I can tailor one of these patterns precisely.