Great question — and yes,

run_deployment

is exactly what you want. Assuming Prefect 3.x: - Use

prefect.deployments.run_deployment

to trigger another flow’s deployment from your parent flow. This runs the child as a separate, orchestrated flow run (not a direct function call). -

run_deployment(...)

returns a

FlowRun

object. To get the child flow’s return value, read it from the

FlowRun.state

using

prefect.states.get_state_result(...)

. Here’s a complete example (sync):

from prefect import flow
from prefect.deployments import run_deployment
from prefect.states import get_state_result, raise_state_exception

@flow
def parent_flow():
    # The name is "<flow_name>/<deployment_name>"
    fr = run_deployment(
        name="child_flow/prod",          # your child deployment
        parameters={"x": 42},            # parameters for the child flow
        as_subflow=True,                 # keep parent/child linkage in UI (default True)
        timeout=600,                     # optional: wait up to 10m
        poll_interval=5,                 # optional: poll every 5s
        tags=["triggered-by-parent"],    # optional: attach tags to the child run
    )

    # Raise if the child failed/crashed
    raise_state_exception(fr.state)

    # Retrieve the actual return value of the child flow
    result = get_state_result(fr.state)
    return result

Notes and best practices: -

run_deployment

blocks until the child run reaches a terminal state (COMPLETED/FAILED/etc.), or until

timeout

is reached. It returns a

FlowRun

object with the final

state

. - Use

get_state_result(fr.state)

to fetch the child flow’s return value. If the child failed, call

raise_state_exception(fr.state)

to surface the error cleanly. -

as_subflow=True

(default) links the child run as a subflow in the UI. Set

as_subflow=False

to decouple it. - You can pass other useful args like

idempotency_key

,

work_queue_name

,

scheduled_time

, and

flow_run_name

. - Async flows:

run_deployment

is sync-compatible. You can

await run_deployment(...)

in an async flow;

get_state_result

is also sync-compatible. Caveats: - The child flow’s return value must be serializable with your configured serializer for Prefect to persist and retrieve it. JSON-serializable values or using the default serializer typically work well. Non-serializable returns will cause

get_state_result

to fail. - If you schedule the child in the future via

scheduled_time

,

run_deployment

will still wait for completion — consider

timeout

accordingly. If you still need to create the child deployment, do it in 3.x with

flow.from_source(...).deploy(...)

(not the removed 2.x commands), for example:

from prefect import flow

@flow
def child_flow(x: int) -> int:
    return x * 2

if __name__ == "__main__":
    child_flow.from_source(
        source="<https://github.com/your-org/your-repo>",
        entrypoint="flows/child.py:child_flow",
    ).deploy(
        name="prod",
        work_pool_name="your-pool",
    )

Docs: - Run deployments: How to run deployments - States and results: States (getting results, raising exceptions) If you’re on Prefect 2.x instead, let me know and I’ll tailor the guidance.

@Marvin MissingResult('State data is missing. Typically, this occurs when result persistence is disabled and the state has been retrieved from the API.')

That error means the child flow’s result was not persisted. You’re calling

get_state_result

on a state that was fetched from the API (as

run_deployment

does), but there’s no persisted payload attached to that state. How to fix it: - Enable result persistence on the CHILD flow (the one you are triggering via

run_deployment

). - Choose a serializer that can handle your return type (JSON for simple types; Pickle for arbitrary Python objects). - Ensure result storage is configured and accessible from where runs execute. Minimal example (child + parent):

from prefect import flow
from prefect.deployments import run_deployment
from prefect.serializers import JSONSerializer, PickleSerializer
from prefect.filesystems import LocalFileSystem
from prefect.states import get_state_result, raise_state_exception

# Configure result storage (local folder, NFS mount, etc.)
results_fs = LocalFileSystem(basepath="/shared/prefect-results")

# Child flow must persist its result so the parent can read it
@flow(
    persist_result=True,
    result_storage=results_fs,
    result_serializer=JSONSerializer(),  # use PickleSerializer() for non-JSON types
)
def child_flow(x: int) -> dict:
    return {"doubled": x * 2}

@flow
def parent_flow():
    fr = run_deployment(
        name="child_flow/prod",
        parameters={"x": 21},
        as_subflow=True,
    )
    # Raise immediately if the child failed
    raise_state_exception(fr.state)

    # Now there's persisted data, so this will succeed
    result = get_state_result(fr.state)
    return result

Tips and caveats: - If your child returns complex objects (e.g., DataFrames), switch to

PickleSerializer()

or

CompressedPickleSerializer()

. - Ensure the configured

result_storage

is reachable by runs. For distributed workers, use a shared path (NFS) or remote/object storage. - If you can’t or don’t want to persist large results, consider returning a small reference (e.g., a URI) and write the heavy payload to storage yourself, or publish an artifact. - This applies to tasks too: if you plan to read a task’s result via the API later, decorate the task with

persist_result=True

and configure storage/serializer similarly. Docs: - States and results: States (getting results) If you can share: - What type does your child return? - Where do your runs execute (local, Docker, K8s), and do you have shared storage available? I can suggest an appropriate

result_serializer

and

result_storage

setup based on that.

Short answer: no. Prefect does not automatically delete persisted results after the run completes or after you read them with

get_state_result

. What happens: - The flow’s final state in the API remains. - The result payload you persisted (e.g., file/object in your configured

result_storage

) stays where it was written until you delete it yourself. Recommended cleanup options: - Object storage lifecycle policies (S3/GCS/Azure) to expire objects under your results prefix after N days. - For local/shared filesystems, schedule a small maintenance flow/cron to prune old result files by age. - If you don’t need to persist large payloads, return a small reference (path/URI) and store the heavy data yourself with your own retention. - Note: If you delete the stored payload, future calls to

get_state_result

for that run will raise

MissingResult

again. If you tell me what storage you’re using (local path, S3, GCS, etc.), I can share a concrete cleanup pattern. Docs on states/results: States and results

That error means you passed an unsaved block instance to

result_storage

. For Prefect to fetch the config on remote workers, the storage block must be saved (persisted server-side) first. Do this: - One-time setup: create and save the block

from prefect.filesystems import LocalFileSystem

# Run once (e.g., locally or in a setup script)
LocalFileSystem(basepath="/shared/prefect-results").save(
    "shared-results",
    overwrite=True,  # set True when updating
)

- In your flow code: load the saved block and pass it to the decorator

from prefect import flow
from prefect.filesystems import LocalFileSystem
from prefect.serializers import JSONSerializer  # or PickleSerializer

results_fs = LocalFileSystem.load("shared-results")

@flow(
    persist_result=True,
    result_storage=results_fs,           # now it's a persisted block
    result_serializer=JSONSerializer(),  # use Pickle for non-JSON types
)
def child_flow(x: int) -> dict:
    return {"doubled": x * 2}

Notes: - You can also create the block in the UI (Blocks -> Filesystems -> Local File System) and then load it in code with

LocalFileSystem.load("your-block-name")

. - For multi-machine runs, prefer a shared location (NFS) or object storage (S3/GCS/Azure) so workers can read/write results. - If you switch to S3/GCS/Azure, the pattern is the same: create the bucket block,

.save(...)

, then

BlockType.load(...)

in your flow and set

persist_result=True

. Related docs: - States and results: States and results (getting results) If you tell me where your workers run and what return type you have, I can recommend a concrete storage + serializer combo.

That means the

LocalFileSystem

block named

shared-results

does not exist in your current workspace (or it was created under a different name/workspace). Fix it with these steps: 1) Verify you’re in the right workspace and see what blocks exist

prefect profile ls
prefect cloud workspace ls   # if using Cloud
prefect block ls

2) Create and save the block (one-time), then verify

from prefect.filesystems import LocalFileSystem

LocalFileSystem(basepath="/shared/prefect-results").save(
    "shared-results",
    overwrite=True,
)

Verify it’s there:

prefect block inspect local-file-system/shared-results

3) Load and use the saved block in your flow

from prefect import flow
from prefect.filesystems import LocalFileSystem
from prefect.serializers import JSONSerializer  # or PickleSerializer

results_fs = LocalFileSystem.load("shared-results")

@flow(persist_result=True, result_storage=results_fs, result_serializer=JSONSerializer())
def child_flow(x: int) -> dict:
    return {"doubled": x * 2}

Notes: - The block name must match exactly (case-sensitive). - If your runs execute on other machines/containers, ensure

/shared/prefect-results

is a shared path all workers can read/write. Otherwise, use object storage (S3/GCS/Azure) with the corresponding block instead. - If you created the block in the UI, double-check you’re in the same workspace and that the block’s name is

shared-results

. Docs: - Blocks (overview): Blocks - States and results: States and results If you share whether you’re on Prefect Cloud vs Server and where your workers run, I can suggest the best storage (local path vs S3/GCS) and a concrete setup.

That error means your flow is still receiving an unsaved block object for

result_storage

at import time, even if you created the block in your workspace. Common causes and fixes: - Constructing a new block instead of loading the saved one - Bad:

results_fs = LocalFileSystem(basepath="/shared/prefect-results")  # unsaved instance
  @flow(persist_result=True, result_storage=results_fs)
  def child(...): ...

- Good:

from prefect.filesystems import LocalFileSystem
  RESULTS = LocalFileSystem.load("shared-results")  # loads the saved block
  @flow(persist_result=True, result_storage=RESULTS)
  def child(...): ...

- Saving after the decorator has already run - The

@flow(...)

decorator is evaluated at import time. If you do

.save()

later in the file, it’s too late. Move block creation/saving to a separate one-time setup script or the UI, then only

load(...)

in your flow module. - Wrong workspace/profile on the worker - Your run may be using a different Prefect profile/workspace than where you created the block. On the machine/container where the worker runs:

prefect cloud workspace ls
  prefect cloud workspace set <org>/<workspace>
  prefect block inspect local-file-system/shared-results

Ensure the block shows up there. - Referencing another unsaved block - If you’re using S3/GCS/etc. storage, ensure any nested credentials blocks are also saved and that the top-level storage block references those saved docs. A known-good pattern end-to-end: 1) One-time: create the block (UI or script)

from prefect.filesystems import LocalFileSystem
LocalFileSystem(basepath="/shared/prefect-results").save("shared-results", overwrite=True)

2) In your flow module, only load and use it

from prefect import flow
from prefect.filesystems import LocalFileSystem
from prefect.serializers import JSONSerializer

RESULTS = LocalFileSystem.load("shared-results")

@flow(persist_result=True, result_storage=RESULTS, result_serializer=JSONSerializer())
def child_flow(x: int) -> dict:
    return {"doubled": x * 2}

3) Verify on the worker:

prefect block inspect local-file-system/shared-results

If you’re still seeing the error, can you share the snippet where you define

result_storage

and the

@flow(...)

decorator, and confirm: - Where the block was created (UI vs code) - Where your worker runs (local, Docker, K8s) - Which workspace the worker is pointed at Docs: - Blocks overview: Blocks - States/results: States and results