Kyle McChesney
03/03/2026, 6:50 PMMarvin
03/03/2026, 6:50 PMMarvin
03/03/2026, 6:53 PMfrom uuid import UUID
from prefect.client.orchestration import get_client
from prefect.client.schemas.filters import ArtifactFilter, ArtifactFilterFlowRunId
from prefect.client.schemas.sorting import ArtifactSort
flow_run_id = UUID("00000000-0000-0000-0000-000000000000")
async def list_artifacts_for_run(flow_run_id: UUID):
async with get_client() as client:
artifacts = await client.read_artifacts(
artifact_filter=ArtifactFilter(
flow_run_id=ArtifactFilterFlowRunId(any_=[flow_run_id])
),
sort=ArtifactSort.UPDATED_DESC,
limit=200,
)
for a in artifacts:
print(a.id, a.key, a.type, a.created, a.updated)
# If you prefer sync:
# with get_client(sync_client=True) as client:
# artifacts = client.read_artifacts(
# artifact_filter=ArtifactFilter(
# flow_run_id=ArtifactFilterFlowRunId(any_=[flow_run_id])
# ),
# limit=200,
# )
Python API (latest version per key only)
from uuid import UUID
from prefect.client.orchestration import get_client
from prefect.client.schemas.filters import (
ArtifactCollectionFilter, ArtifactCollectionFilterFlowRunId
)
flow_run_id = UUID("00000000-0000-0000-0000-000000000000")
async def list_latest_artifacts_for_run(flow_run_id: UUID):
async with get_client() as client:
latest = await client.read_latest_artifacts(
artifact_filter=ArtifactCollectionFilter(
flow_run_id=ArtifactCollectionFilterFlowRunId(any_=[flow_run_id])
),
limit=200,
)
for col in latest:
# col.key is the artifact key; col.latest_id is the latest artifact version id
print(col.key, col.latest_id)
Alternative filter (via FlowRunFilter)
from prefect.client.schemas.filters import FlowRunFilter, FlowRunFilterId
async with get_client() as client:
artifacts = await client.read_artifacts(
flow_run_filter=FlowRunFilter(id=FlowRunFilterId(any_=[flow_run_id])),
limit=200,
)
REST API
- Endpoint: POST /artifacts/filter (against your Prefect API)
- Body:
{
"artifacts": {
"flow_run_id": {
"any_": [
"00000000-0000-0000-0000-000000000000"
]
}
},
"limit": 200
}
CLI
- There isn’t a flag to filter by flow run id. prefect artifact ls lists artifacts, but has no --flow-run-id option:
prefect artifact ls --help
You’ll need the Python client or REST API for run-scoped queries.
If you’re on Prefect 2.x or need a ready-to-run helper script, let me know and I’ll tailor it.