Syméon del Marmol
10/23/2025, 9:54 AMMarvin
10/23/2025, 9:54 AMMarvin
10/23/2025, 10:01 AMcreate_markdown_artifact(...) do not accept run IDs; use the artifact classes directly and call `.create()`/`.acreate()`.
Example (async) that looks up a task run by name within a flow run, then creates a Markdown artifact attached to that task run:
from uuid import UUID
from prefect.client.orchestration import PrefectClient
from prefect.client.schemas.filters import TaskRunFilter, TaskRunFilterFlowRunId, TaskRunFilterName
from prefect.artifacts import MarkdownArtifact
async def attach_markdown_to_task(flow_run_id: UUID, task_name: str, markdown: str, key: str | None = None):
async with PrefectClient() as client:
task_runs = await client.read_task_runs(
task_run_filter=TaskRunFilter(
flow_run_id=TaskRunFilterFlowRunId(any_=[flow_run_id]),
name=TaskRunFilterName(any_=[task_name]),
)
)
if not task_runs:
raise RuntimeError(f"No task runs named {task_name!r} found in flow run {flow_run_id}")
task_run_id = task_runs[0].id
artifact = MarkdownArtifact(
markdown=markdown,
key=key, # optional; if provided, must be lowercase letters, numbers, and dashes
task_run_id=task_run_id,
flow_run_id=flow_run_id,
)
artifact_id = await artifact.acreate()
return artifact_id
Notes:
- If you are already inside the target task run (e.g., inside the function executed via .delay() on a task worker), just call the helpers like create_markdown_artifact(...) — Prefect will infer the run IDs from context.
- Creating artifacts entirely “unattached” (no run context and no IDs) is discouraged; you’ll see a warning. If you must create them from your backend, pass the task_run_id (and flow_run_id) as shown above.
- Since you dynamically name tasks with .with_options(name=...), filtering by TaskRunFilterName will match that name.
Docs:
- Artifacts guide
- prefect.artifacts Python API
- Filters API
- Background tasks
If you’d like, I can tailor this to your exact pattern (e.g., your backend initiating artifacts after asyncio.gather) or provide a sync version.