Tom Han
08/29/2025, 6:38 AMMarvin
08/29/2025, 6:38 AMMarvin
08/29/2025, 6:45 AMfrom prefect import flow
from prefect.artifacts import create_markdown_artifact
from prefect.runtime import flow_run
@flow
def child():
artifact_id = create_markdown_artifact(
markdown="Hello from subflow!",
key=f"report-{flow_run.id}" # optional but helpful
)
return {"artifact_id": str(artifact_id), "run_id": flow_run.id}
Parent:
from prefect import flow
from prefect.client.orchestration import SyncPrefectClient
@flow
def parent():
info = child()
with SyncPrefectClient() as client:
art = client.read_artifact(info["artifact_id"])
# 'art' is an Artifact model; inspect art.type, art.key, art.data, etc.
return art
Pattern B — Use a predictable artifact key and query for it from the parent
- Use the key
field when creating the artifact in the subflow, then look it up by key from the parent. Keys also form “collections” so you can always fetch the latest for a given key.
Subflow:
from prefect import flow
from prefect.artifacts import create_markdown_artifact
from prefect.runtime import flow_run
@flow
def child():
key = f"daily-report-{flow_run.id}" # or another stable naming convention
create_markdown_artifact(markdown="# Subflow Report", key=key)
return {"key": key}
Parent:
from prefect import flow
from prefect.client.orchestration import SyncPrefectClient
from prefect.client.schemas.filters import ArtifactFilterKey
@flow
def parent():
info = child()
with SyncPrefectClient() as client:
artifacts = client.read_artifacts(
artifact_filter=ArtifactFilterKey(any_=[info["key"]])
)
# 'artifacts' is a list; typically you'll want the latest one for that key
return artifacts[0] if artifacts else None
Optional: Filter by the subflow run id
- If you also return the subflow’s run id (or otherwise know it), you can list all artifacts created by that subflow run:
from prefect.client.orchestration import SyncPrefectClient
from prefect.client.schemas.filters import ArtifactFilterFlowRunId
with SyncPrefectClient() as client:
artifacts = client.read_artifacts(
artifact_filter=ArtifactFilterFlowRunId(any_=[subflow_run_id])
)
Notes and tips
- There is no automatic way to “attach” a subflow’s artifact to the parent run; you fetch it via the API using the artifact’s ID, key, or the subflow’s run id.
- Keys are great for decoupling: the parent doesn’t need to know the subflow’s run id if it can rely on a known key.
- If you want the parent to surface a summary, a common pattern is to create a link artifact in the parent that points to the subflow artifact’s UI URL, or to copy essential content from the subflow artifact into a parent artifact.
Helpful docs
- Artifacts API (creation helpers): Artifacts API
- Python client (reading artifacts): Python client
- Filters for queries (e.g., ArtifactFilterKey, ArtifactFilterFlowRunId): Filters API
If you can share how you’re creating the artifact (type and whether you can set a key) and whether you prefer sync or async code, I can tailor an example precisely to your setup.