Yes — in Prefect 3.x you can save metadata during a flow run, but how you do it depends on what you mean by “metadata.” Best options: - Artifacts → for progress, reports, tables, links, summaries - Events → for structured runtime signals - Tags / labels → for lightweight run metadata - Flow run context → to read run info during execution A few important details: 1. Read current flow-run metadata inside the flow

python
from prefect import flow
from prefect.context import get_run_context

@flow
def my_flow():
    ctx = get_run_context()
    flow_run = ctx.flow_run

    print(flow_run.id)
    print(flow_run.name)
    print(flow_run.tags)
    print(flow_run.parameters)
    print(flow_run.labels)

2. Save metadata during the run using artifacts This is usually the most user-friendly approach because artifacts show up in the UI.

python
from prefect import flow
from prefect.artifacts import create_markdown_artifact, create_table_artifact

@flow
def my_flow():
    create_markdown_artifact(
        key="run-summary",
        markdown="""
# Flow metadata
- source: api
- batch_id: 12345
- status: processing
""",
        description="Metadata collected during the flow run"
    )

    create_table_artifact(
        key="metrics",
        table=[
            {"name": "rows_processed", "value": 1000},
            {"name": "errors", "value": 3},
        ],
    )

3. Save structured runtime metadata with events If you want machine-readable metadata for observability, emit events.

python
from prefect import flow
from prefect.events import emit_event

@flow
def my_flow():
    emit_event(
        event="my-flow.metadata",
        resource={"prefect.resource.id": "custom-resource"},
        payload={
            "batch_id": "12345",
            "rows_processed": 1000,
            "stage": "transform",
        },
    )

4. Update labels on a flow run at runtime If you specifically want metadata attached to the flow run record itself, labels are one option.

python
from prefect import flow, get_client
from prefect.context import get_run_context

@flow
def my_flow():
    flow_run_id = get_run_context().flow_run.id

    with get_client(sync_client=True) as client:
        client.update_flow_run_labels(
            flow_run_id,
            {
                "dataset": "customers",
                "region": "us-east-1",
                "batch_id": "12345",
            },
        )

Limitations - The in-memory

flow_run

object is not something you just mutate directly and have persist. - Runtime updates to tags/labels are done through the API client. - If you need rich metadata/history during execution, artifacts are usually better than trying to stuff everything into the flow-run record. Recommendation - For “I want to record metadata during execution and see it later” → use artifacts - For “I want structured operational signals” → use events - For “I want small key/value metadata attached to the run” → use labels If you want, I can give you: 1. a minimal example of attaching metadata to the current flow run, or 2. a pattern for storing per-step metadata throughout execution.

Got it — if you want a global mutable parameter that every task can read and update during the flow run, and then inspect later after the run, Prefect does not have a built-in “shared global flow variable” for that. Why: - Prefect tasks may run in different processes / workers / infrastructure - so normal Python global variables or shared in-memory objects are not reliable - task execution is designed to be distributed, so shared mutable state inside the flow is not a core Prefect primitive Recommended patterns instead: Option 1: Use an external store (best for true shared mutable state) Have every task read/write to something external: - database row - Redis - S3 / GCS / blob storage JSON file - key-value store Example idea:

python
from prefect import flow, task

global_state = {}

@task
def write_state(key, value):
    global global_state
    global_state[key] = value

@task
def read_state(key):
    return global_state.get(key)

But do not rely on the above except in purely local, single-process execution. In real Prefect deployments, use something external instead:

python
from prefect import flow, task
import json
from pathlib import Path

STATE_FILE = Path("/tmp/flow_state.json")

def load_state():
    if STATE_FILE.exists():
        return json.loads(STATE_FILE.read_text())
    return {}

def save_state(state):
    STATE_FILE.write_text(json.dumps(state))

@task
def update_state(key, value):
    state = load_state()
    state[key] = value
    save_state(state)

@task
def get_state(key):
    state = load_state()
    return state.get(key)

@flow
def my_flow():
    update_state("status", "running")
    print(get_state("status"))

That demonstrates the pattern, but for production you’d usually use Redis/DB/object storage, not

/tmp

. Option 2: Pass state explicitly between tasks If tasks do not need arbitrary shared mutation, the cleanest Prefect-native pattern is to pass a dictionary/result from one task to another.

python
from prefect import flow, task

@task
def init_state():
    return {"status": "started", "count": 0}

@task
def update_state(state):
    state["count"] += 1
    state["status"] = "processing"
    return state

@task
def finalize(state):
    state["status"] = "done"
    return state

@flow
def my_flow():
    state = init_state()
    state = update_state(state)
    state = finalize(state)
    print(state)

This works well, but it is not a shared global variable — it is explicit dataflow. Option 3: Persist metadata snapshots as artifacts (best for “see it later in UI”) If your main goal is “tasks update metadata and I want to inspect it after the run”, then artifacts are a strong fit. Each task can publish a new snapshot.

python
from prefect import flow, task
from prefect.artifacts import create_markdown_artifact

@task
def step_one():
    create_markdown_artifact(
        key="shared-state",
        markdown="status: step_one_complete\ncount: 1"
    )

@task
def step_two():
    create_markdown_artifact(
        key="shared-state",
        markdown="status: step_two_complete\ncount: 2"
    )

@flow
def my_flow():
    step_one()
    step_two()

This is good for observability, but artifacts are not really a concurrency-safe shared mutable object for tasks to coordinate through. What I’d recommend for your use case If you need all tasks to read/update the same state during execution and inspect it later: - use an external persisted store as the source of truth - optionally also write a final artifact summary for easy viewing in Prefect UI A very practical pattern is: - Redis/DB/object-store JSON = mutable shared state - Prefect artifact = final human-readable summary Important caveat If tasks run in parallel, then multiple tasks updating the same global state can conflict. You may need: - locking - atomic updates - versioning - merge logic So the right backend depends on your concurrency needs. Short answer:

Prefect does not provide a built-in mutable global flow parameter shared across tasks. Use an external persisted store for shared state, and artifacts if you want to view the final metadata later in the UI. If you want, I can show you a complete example using: - Redis - a JSON file in object storage - a database row - or a Prefect-friendly pattern for local-only execution.