Tate Wilks
10/31/2025, 8:00 PMdeploy_args = {
# fmt: off
"name" : name,
"work_pool_name" : "workers-k8s",
"image" : image,
"parameters" : parameters,
"description" : description,
"tags" : tags,
# fmt : on
}
flow.deploy(**deploy_args)Marvin
10/31/2025, 8:00 PMMarvin
10/31/2025, 8:07 PMtags you pass to flow.deploy(...) are deployment tags, not flow run tags—so runs created by that deployment will not inherit them. This is expected behavior.
What to use instead if you want tags on your flow runs:
- Add tags in code at runtime with the tags context manager. This applies to every run, including scheduled ones.
from prefect import flow, tags
@flow
def my_flow():
with tags("prod", "etl"):
# your flow logic
...
If you want them to be dynamic:
from prefect import flow, tags
@flow
def my_flow(run_tags: list[str] | None = None):
with tags(*(run_tags or [])):
...
- When starting a run programmatically, pass tags to `run_deployment`:
from prefect.deployments import run_deployment
run = run_deployment("my-flow/my-deployment", tags=["prod", "etl"])
Notes:
- Deployment tags are useful for organizing/filtering deployments in the UI and automations; they’re not used by workers and are not inherited by runs.
- Concurrency limits and run-based filtering target flow run tags, so you need to add tags at run time (via the context manager or when creating the run).
If you can share what you’re using tags for (e.g., concurrency limits, searching in the UI, etc.), I can suggest the cleanest pattern for your setup.