Servando
03/20/2026, 12:31 PMMarvin
03/20/2026, 12:31 PMMarvin
03/20/2026, 12:39 PMflow-run-feedback-<run_id>, is solid. Keys must be lowercase letters/numbers/dashes only; UUID run IDs fit that.
- Artifacts (except progress) are immutable. If you’ll receive multiple UI updates for the same run, write a full snapshot each time with the same key — Prefect will keep versions and the “latest” will be returned by lookups. This works well for low volume feedback.
Schema and artifact type
- Prefer create_table_artifact for structured annotations. Use a list of dicts (one row per label) with consistent columns like item_id, label, annotator, timestamp, notes, etc.
- If your feedback is free-form, use create_markdown_artifact and render a compact report.
- For arbitrary JSON that doesn’t fit table/markdown, you can use the Artifact class directly with data=... and a key (see code pattern below).
Discoverability and linkage to runs
- If you create the artifact from inside a flow/task, it’s automatically linked to that run and shows up on the run’s Artifacts tab.
- If you’re creating it outside of a run (e.g., from a UI service), use a key that includes the run_id like you proposed; you can still fetch by key. If you also want it visibly attached to the run in the UI, create it from a small flow (triggered via webhook) so the run context is present.
Recommended patterns
- Single-snapshot per run (best if feedback is final/rarely updated)
from prefect.artifacts import create_table_artifact
def write_feedback(run_id: str, rows: list[dict]):
# rows example: [{"item_id": "x1", "label": "approved", "annotator":"sam", "timestamp":"..."}, ...]
create_table_artifact(
table=rows,
key=f"flow-run-feedback-{run_id}",
description="# Human feedback for this run"
)
- Incremental updates (replace-as-new-version)
from prefect.artifacts import create_table_artifact
def upsert_feedback(run_id: str, all_rows: list[dict]):
# Always send the full current set; this will create a new artifact version
create_table_artifact(
table=all_rows,
key=f"flow-run-feedback-{run_id}",
description="Updated feedback snapshot"
)
- Retrieve latest for a run
from prefect.artifacts import Artifact
artifact = Artifact.get(key=f"flow-run-feedback-{run_id}")
if artifact:
data = artifact.data # For table artifacts, this is the normalized table structure
- Generic JSON artifact (when table/markdown aren’t a good fit)
from prefect.artifacts import Artifact
artifact, created = Artifact.get_or_create(
key=f"flow-run-feedback-{run_id}",
description="Human feedback (JSON)",
data={"labels": rows, "meta": {"annotators": ["sam", "lee"]}}
)
# Note: non-progress artifacts are immutable; to “update”, create another artifact with the same key.
Operational tips
- Keep payloads small. Artifacts are best for small/medium JSON or table payloads; don’t store large blobs/binaries — put those in object storage and link with create_link_artifact.
- Add context in description (annotator(s), version, UI build) to make the UI more useful.
- Namespacing: if you need multiple feedback sets per run, either version a single key (recommended) or suffix the key, e.g., flow-run-feedback-<run_id>-qa.
- Retention: artifacts follow your plan’s data retention. If you need to keep labels longer than your plan allows, export them periodically to external storage.
- Security: don’t call the Prefect API directly from the browser with a workspace key. Use a tiny backend/service (or a short-lived “ingest” flow run triggered via webhook) to write artifacts.
Docs and API references
- Concepts: Artifacts conceptsMarvin
03/20/2026, 12:39 PMServando
03/20/2026, 1:16 PMMarvin
03/20/2026, 1:16 PMMarvin
03/20/2026, 1:21 PMflow_run_id (and optionally task_run_id). Use your UI’s backend/service with a Prefect API key to do this.
Recommended pattern (attach to a completed flow run)
- Use PrefectClient.create_artifact(...) with ArtifactCreate and pass:
- `key`: e.g., flow-run-feedback-<run_id> (lowercase/numbers/dashes)
- `type`: "table", "markdown", "link", or "image"
- `data`: your annotations payload (table or markdown are most common)
- `flow_run_id`: the UUID of the completed run
Python (sync) example for post-run annotations
from uuid import UUID
from prefect.client.orchestration import get_client
from prefect.client.schemas.actions import ArtifactCreate
def store_post_run_feedback(flow_run_id: str, rows: list[dict]):
# rows example: [{"item_id": "x1", "label": "approved", "annotator":"sam", "timestamp":"..."}, ...]
key = f"flow-run-feedback-{flow_run_id}"
with get_client(sync_client=True) as client:
artifact = ArtifactCreate(
key=key,
type="table", # or "markdown"
description="Human feedback (post-run)",
data=rows, # list of dicts, or use dict-of-lists if you prefer
flow_run_id=UUID(flow_run_id),
)
created = client.create_artifact(artifact)
return created
cURL (direct REST) example
curl -X POST "<https://api.prefect.cloud/api/accounts/<account_id>/workspaces/<workspace_id>/artifacts/>" \
-H "Authorization: Bearer <PREFECT_API_KEY>" \
-H "Content-Type: application/json" \
-d '{
"key": "flow-run-feedback-<run_id>",
"type": "table",
"description": "Human feedback (post-run)",
"data": [
{"item_id":"x1","label":"approved","annotator":"sam","timestamp":"2026-03-20T10:12:00Z"},
{"item_id":"x2","label":"rework","annotator":"lee","timestamp":"2026-03-20T10:13:10Z"}
],
"flow_run_id": "<run_uuid>"
}'
After creation
- The artifact will:
- Appear on the flow run’s Artifacts tab (because you set flow_run_id)
- Be globally discoverable by CLI and API using its key
- List and inspect via CLI:
- prefect artifact ls
prefect artifact inspect flow-run-feedback-<run_id>
- Programmatic lookup (latest by key):
- from prefect.artifacts import Artifact
art = Artifact.get(key=f"flow-run-feedback-{run_id}")
A few tips
- If you expect multiple updates for the same run, write a new artifact with the same key each time. Prefect versions artifacts; inspect will show all versions and Artifact.get(key=...) returns the latest.
- Keep payloads reasonably small; for large blobs, store in object storage and link with create_link_artifact.
- Don’t call the Prefect API from the browser; send from your backend using a workspace API key.
- Retention follows your Prefect Cloud plan.
Docs for reference
- Concepts: Artifacts concepts
- How-to: How to create and use artifacts
- Python API: prefect.artifacts API
- CLI: prefect artifact --help
If you share how your UI batches updates (one final snapshot vs. many increments) and your expected payload shape/size, I can tailor the artifact schema and keying/versioning pattern further.