Thanks for the snippet — a couple of clarifications and a working pattern you can use. What run_deployment returns and how to get the result - In Prefect 3.x,

prefect.deployments.run_deployment(...)

returns a

FlowRun

object. - By default it waits for the run to complete (you can change this with

timeout

, see below). - To get the child flow’s return value, call

await flow_run.state.result()

once it’s in a terminal state. If the run failed,

result()

will raise unless you pass

raise_on_failure=False

. Notes: - In an async flow, it’s correct to

await run_deployment(...)

. - Don’t call

state.result()

until the state is terminal (COMPLETED/FAILED/etc.); otherwise you’ll get an error because there’s no result to resolve yet. Fixed version of your pattern

from typing import Optional, Tuple
from prefect import flow
from prefect.deployments import run_deployment
from prefect.client.orchestration import get_client

async def get_flow_run_result(flow_run_id):
    async with get_client() as client:
        flow_run = await client.read_flow_run(flow_run_id)
        if not flow_run.state.is_final():
            raise RuntimeError(f"Flow run {flow_run_id} not finished (state={flow_run.state.type})")
        return await flow_run.state.result()

@flow(log_prints=True)
async def s2master_flow(
    ndays: Optional[int] = 20,
    ndays_field: Tuple[str, ...] = ("PublicationDate",),
    tile: Tuple[str, ...] = (),
    esa_id: Tuple[str, ...] = (),
    odata_filter: Optional[str] = None,
    env_state: str = "dev",
):
    main_params = {
        "ndays": ndays,
        "ndays_field": ndays_field,
        "tile": tile,
        "esa_id": esa_id,
        "odata_filter": odata_filter,
    }

    # Waits for completion by default; returns a FlowRun with its final state
    child_run = await run_deployment(
        name=f"s2-import/s2import_{env_state}",
        parameters=main_params,
        # as_subflow=True  # default when called from within a flow; links runs in the UI
        # timeout=None     # default = wait until completion; set timeout=0 for non-blocking
    )

    if child_run.state.is_completed():
        value = await child_run.state.result()
        print(f"Child flow result: {value}")
    else:
        # Get the error object without raising (if FAILED/CRASHED)
        err_or_val = await child_run.state.result(raise_on_failure=False)
        print(f"Child flow finished in state {child_run.state.type}: {err_or_val}")

Non-blocking variant

child_run = await run_deployment(
    name=f"s2-import/s2import_{env_state}",
    parameters=main_params,
    timeout=0,  # return immediately
)

# ... do other work ...

# Later, poll for completion and fetch the result
async with get_client() as client:
    while True:
        fr = await client.read_flow_run(child_run.id)
        if fr.state.is_final():
            break
        await anyio.sleep(2)

    value = await fr.state.result()  # raises if FAILED unless raise_on_failure=False

Subflow vs run_deployment (when to use which) - Call another flow directly (subflow) if you control both flows, want in-process execution, and don’t need independent scheduling/cancellation:

@flow
  async def child(...): ...
  @flow
  async def parent(...):
      result = await child(...)

- Use

run_deployment

when you want an independently orchestrated run with its own work pool/schedule or you may need to cancel/observe it separately. When called inside a flow, it’s linked as a subflow in the UI by default (you can opt out with

as_subflow=False

). Common gotcha: result persistence - If

state.result()

raises a “missing result” error, configure result persistence for the child flow/deployment (e.g., set

persist_result=True

on the flow/task or configure result storage on the deployment). Without persistence, large/complex return values may not be retrievable remotely.

If you’d like, share any error you’re hitting when calling

state.result()

and I can help pinpoint whether it’s a timing/state issue or a result persistence configuration issue.

Thanks for the detailed trace — this helps a lot. What’s going on - You’re on a mismatched version set: client v3.4.1 talking to server v3.6.22. - The 422 is coming from the server’s /api/task_runs/ endpoint that’s hit when

run_deployment

is called as a subflow (the default when invoked inside a flow). - The error detail (“Input should be a valid dictionary… input: '{\"state\": ... }'”) indicates the server is receiving the body as a string instead of a JSON object — a serialization/validation mismatch between those client/server versions. In short: this is a client/server compatibility issue during the “create subflow task run” call. Recommended fixes (in order) 1) Upgrade your Prefect client to match the server - Upgrade the environment that runs

s2master_flow

(and your workers) to the same Prefect version as the server, e.g.:

pip install -U "prefect==3.6.22"

Keeping client and server on the same minor/patch version is the safest path. 2) Immediate workaround: disable subflow linking for this call - This avoids creating the “subflow task run” (and therefore avoids the /api/task_runs/ call that’s failing), while still triggering the deployment and returning a FlowRun you can await/poll.

flow_run = await run_deployment(
      name=f"s2-import/s2import_{env_state}",
      parameters=main_params,
      as_subflow=False  # workaround: skip creating the subflow task run
  )
  files_to_process = await (await get_client()).read_flow_run(flow_run.id)  # or use your helper
  files_to_process = await flow_run.state.result()  # once in a terminal state

A cleaned-up version of your flow with the workaround

from typing import Optional, Tuple
from prefect import flow
from prefect.deployments import run_deployment
from prefect.client.orchestration import get_client

async def get_flow_run_result(flow_run_id):
    async with get_client() as client:
        flow_run = await client.read_flow_run(flow_run_id)
        if not flow_run.state.is_final():
            raise RuntimeError(f"Flow run {flow_run_id} not finished (state={flow_run.state.type})")
        return await flow_run.state.result()

@flow(log_prints=True)
async def s2master_flow(
    ndays: Optional[int] = 20,
    ndays_field: Tuple[str, ...] = ("PublicationDate",),
    tile: Tuple[str, ...] = (),
    esa_id: Tuple[str, ...] = (),
    odata_filter: Optional[str] = None,
    env_state: str = "dev",
):
    main_params = {
        "ndays": ndays,
        "ndays_field": ndays_field,
        "tile": tile,
        "esa_id": esa_id,
        "odata_filter": odata_filter,
    }

    # Workaround: don't create a subflow task run (bypasses the failing endpoint)
    child_run = await run_deployment(
        name=f"s2-import/s2import_{env_state}",
        parameters=main_params,
        as_subflow=False
    )

    # Wait for completion and fetch result
    files_to_process = await get_flow_run_result(child_run.id)
    return files_to_process

A couple of additional notes - If you rely on the UI parent/child nesting, re-enable it after upgrading the client by removing

as_subflow=False

(the default is True when called inside a flow). - If

state.result()

ever complains about a missing result, ensure the child flow/deployment has result persistence configured (e.g.,

persist_result=True

and result storage set). References - API reference for `run_deployment`: https://docs.prefect.io/latest/api-ref/prefect/deployments/#prefect.deployments.run_deployment - Subflows and

as_subflow

behavior: https://docs.prefect.io/latest/concepts/flows/#subflows If upgrading the client doesn’t resolve it, let me know: - The exact versions running in the parent flow environment and on your workers - Whether there’s any proxy or gateway in front of the API - A redacted server log for the failing POST /api/task_runs/ request I can dig in further with those details.