<@ULVA73B9P> Is there a way that I can access the ...
# ask-marvin
t
@Marvin Is there a way that I can access the flow run context and modify it with custom values? Specifically, I want to add some custom data to the data returned when querying the server API at /api/flow_run/{id}.
m
thought for 723.5 seconds
Short answer: you can’t mutate the in-memory flow run context to persist arbitrary fields on the FlowRun record. In Prefect 3.x the FlowRun schema is fixed. However, there is a supported way to attach custom key–value data that will show up in GET /api/flow_runs/{id}: use flow run labels. For richer data, use artifacts. What works today - Flow run labels (recommended for small key–value metadata) - Shows up on the flow run at
labels
in the API response - Updateable during a run via the Prefect client - Tags (good for filtering/categorization) - Shows up at
tags
in the API response - Artifacts (best for structured or larger data, files, tables, links) - Not embedded in the flow run JSON, but linked to the run and queryable in the Artifacts API/UI Example: add custom labels from within a flow
Copy code
from prefect import flow
from prefect.context import get_run_context
from prefect.client.orchestration import get_client

@flow
def my_flow():
    ctx = get_run_context()
    flow_run_id = ctx.flow_run.id

    # Sync client for simplicity inside a flow
    with get_client(sync_client=True) as client:
        client.update_flow_run_labels(
            flow_run_id,
            {
                "team": "data-eng",
                "run_type": "backfill",
                "ticket": "ABC-123",
            },
        )

    # Optional: read the run back to confirm
    with get_client(sync_client=True) as client:
        fr = client.read_flow_run(flow_run_id)
        print(fr.labels)  # {'team': 'data-eng', 'run_type': 'backfill', 'ticket': 'ABC-123'}
Read via REST to see your labels
Copy code
curl -H "Authorization: Bearer $PREFECT_API_KEY" \
  "$PREFECT_API_URL/api/flow_runs/<flow_run_id>"
# ...
# "labels": {"team": "data-eng", "run_type": "backfill", "ticket": "ABC-123"},
# ...
Example: add tags (simple flags you can filter by)
Copy code
from prefect import flow, tags

@flow
def my_flow():
    with tags("backfill", "team:data-eng"):
        # tasks here inherit these tags; the flow run will have them too
        pass
Example: attach richer metadata as an artifact
Copy code
from prefect import flow
from prefect.artifacts import create_markdown_artifact, create_table_artifact

@flow
def my_flow():
    # Simple markdown summary
    create_markdown_artifact(
        markdown="# Run summary\n- team: data-eng\n- run_type: backfill",
        key="run-summary",
        description="Summary of this flow run",
    )

    # Tabular details
    create_table_artifact(
        table=[{"metric": "rows_processed", "value": 12345}],
        key="metrics",
        description="Run metrics",
    )
Notes and guidance - The run context you access with
get_run_context()
is for reading execution info; mutating it won’t persist to the API. - Labels are ideal for small scalar values (strings/numbers/booleans). Keep them concise; for larger or nested data, prefer artifacts. - Job variables appear on the run, but are meant for worker/deployment configuration rather than storing custom run metadata. Relevant references - REST: Update flow run labels docs: update_flow_run_labels - REST: Read a flow run docs: read_flow_run - Artifacts (Python API) docs: prefect.artifacts If you can share what shape/size of data you want to add and how you intend to query it, I can recommend the best mix (labels vs. tags vs. artifacts) and provide a tailored snippet.
n
yea artifacts or results would be the right away to persist arbitrary extra metadata with a flow/task run
t
@Marvin is there a way to clear the labels once they have been updated?
m
thought for 635.1 seconds
t
@Nate to persist results that makes sense, what I'm trying to do is add some kind of context/metadata that will show what step a task is on when hitting the /flow_run/{id} endpoint so I can see the status. If there's a way to see what task is currently running, then that would suit my purpose as well.
m
Good question — do you mean: - Flow run key/value labels - Deployment/work pool “job” labels (e.g., pod/container labels) - Or tags (list of strings) The way to “clear” them depends on which one you’re working with. Here are the common cases: - Flow run labels (key/value) - Yes — you can clear them by updating the run’s labels to an empty dict. - Example:
Copy code
from prefect.client.orchestration import get_client

    flow_run_id = "YOUR-FLOW-RUN-ID"

    async with get_client() as client:
        # Replace the current label set with an empty dict
        await client.update_flow_run_labels(flow_run_id, labels={})
- API ref: API: Update flow run labels - Tags (list of strings) on a flow run - Clear by setting tags to an empty list:
Copy code
from prefect.client.orchestration import get_client
    from prefect.client.schemas.actions import FlowRunUpdate

    async with get_client() as client:
        await client.update_flow_run(flow_run_id, FlowRunUpdate(tags=[]))
- More on the client: Python API client - Worker/job labels (e.g., Kubernetes/Docker labels applied by the worker) - Clear by updating your work pool base job template or deployment
job_variables
to set
labels
to `{}`:
Copy code
flow.from_source("<https://github.com/owner/repo>", entrypoint="path/to/flow.py:my_flow").deploy(
        name="my-deploy",
        work_pool_name="my-pool",
        job_variables={"labels": {}},  # clears labels on the created job/pod
    )
- Background on customizing workers: Developing a custom worker Notes: - In Prefect 3.x, label updates for flow runs are applied to the server’s label map. Passing
{}
removes all labels. If you still see old keys after doing this, let me know your Prefect version and exactly which method you’re calling and I’ll dig in. - If you meant a different resource, tell me which one and I’ll share the exact steps.
t
In Prefect 3.x, label updates for flow runs are applied to the server’s label map. Passing
{}
removes all labels. If you still see old keys after doing this, let me know your Prefect version and exactly which method you’re calling and I’ll dig in.
I tried running
client.update_flow_run_labels(flow_run_id, {})
but the old keys persisted. I'm running prefect 3.4.7 @Marvin
m
thought for 327.8 seconds
Thanks for the detail — and thanks for catching that. On Prefect 3.4.7,
update_flow_run_labels
uses merge semantics. Sending
{}
is effectively a no-op; it won’t remove existing keys. Sorry for the confusion earlier. What works instead is explicitly deleting keys by setting them to `None`: - Remove specific keys:
Copy code
from prefect.client.orchestration import get_client

async with get_client() as client:
    await client.update_flow_run_labels(flow_run_id, {
        "env": None,
        "team": None,
    })
- Remove all existing keys:
Copy code
from prefect.client.orchestration import get_client

async with get_client() as client:
    fr = await client.read_flow_run(flow_run_id)
    current = fr.labels or {}
    if current:
        delete_payload = {k: None for k in current.keys()}
        await client.update_flow_run_labels(flow_run_id, delete_payload)
Relevant API docs: Update flow run labels A couple of notes: - If any labels reappear after removal, they may be system-generated at run time. If you share an example of the keys that persist, I can confirm whether they’re regenerated by the server/worker. - If you prefer a direct REST call, you can PATCH the endpoint with nulls:
Copy code
PATCH {PREFECT_API_URL}/flow-runs/{FLOW_RUN_ID}/labels
  {"labels": {"env": null, "team": null}}
If the
None
removal doesn’t work in your environment, let me know the exact keys that stick around and I’ll dig deeper (and file a bug if needed).