Short answer: in Prefect 3.x there isn’t a mutable “shared context” for arbitrary user data. The context objects you can access from flows/tasks are read-only and intended for execution metadata. The best practice for sharing data between tasks is to pass return values (Futures) explicitly; Pydantic models work great for this. What to use instead of a “context” - Pass results between tasks - Calling a task returns a PrefectFuture; pass that future into downstream tasks. Prefect will resolve it for you. - Works naturally with Pydantic models. - Use a single “context model” you enrich through the pipeline - Define a Pydantic model that represents your working state; each task accepts and returns it. - Persist results when needed - If tasks may run on different machines (e.g., with task workers) or you want durability/caching across retries, enable

persist_result

and choose storage + serializer. Concrete example with a Pydantic “context” model

from pydantic import BaseModel
from prefect import flow, task

class RunContext(BaseModel):
    user_id: int | None = None
    name: str | None = None
    tags: list[str] = []

@task
def fetch_user() -> RunContext:
    return RunContext(user_id=123, name="alice", tags=["seed"])

@task
def enrich(ctx: RunContext) -> RunContext:
    ctx.tags.append("enriched")
    return ctx

@task
def finalize(ctx: RunContext) -> str:
    return f"User {ctx.user_id} - {ctx.name} - {ctx.tags}"

@flow
def pipeline():
    ctx = fetch_user()          # PrefectFuture[RunContext]
    ctx = enrich(ctx)           # pass the future; Prefect resolves it
    msg = finalize(ctx)
    return msg.result()         # only call .result() in flows if you need the value

Result persistence and serialization (when you need durability) - Defaults in 3.x: - Results are not persisted by default. - In-memory caching is on by default for the duration of the flow run. - Enable persistence per task or flow:

@task(
    persist_result=True,
    result_serializer="pickle",        # or "json", "compressed/pickle", etc.
    result_storage="/tmp/prefect-results"  # local path, or a block name like "my-s3-block"
)
def produce_model() -> RunContext:
    ...

- Enable globally (optional): -

PREFECT_RESULTS_PERSIST_BY_DEFAULT=true

in your environment/profile. Notes and gotchas - Context is not a store:

prefect.context.get_run_context()

is for metadata (IDs, parameters, logger, etc.). It’s read-only and not meant for user data. - Don’t stash data in globals or module-level variables: unsafe in distributed/executors and across retries. - Large objects: consider persisting to external storage (S3/GCS/local path) and passing a reference (key/path) instead of the whole object. - Parallel tasks can share the same upstream result by passing the same future to both. If you share how you plan to run the tasks (all in-process vs with task workers / different infrastructure) and the size/type of the data (e.g., big DataFrames vs small Pydantic models), I can suggest an exact pattern and the right storage/serializer settings.