Sorry, I encountered an error while processing your request. Please try again.

You can do this in Prefect 3.x by triggering the child flow’s deployment from within your parent flow using

run_deployment

. This keeps the parent/child (subflow) relationship while executing the child on its own deployment/infra. Key points: - Call

run_deployment("child-flow-name/child-deployment-name", ...)

from inside your parent flow - By default

as_subflow=True

, so the child run is linked as a subflow of the parent - You can route it to different infra via the child deployment itself,

work_queue_name

, or

job_variables

- You can choose to wait for completion or fire-and-forget Minimal examples - Synchronous parent flow:

from prefect import flow
from prefect.deployments import run_deployment

@flow
def parent_flow():
    # Child runs on its own deployment/infra, but is linked as a subflow
    child_run = run_deployment(
        name="child-flow/production",          # flow-name/deployment-name
        parameters={"param1": "value"},
        # wait behavior:
        # timeout=None (default): wait until terminal state
        # timeout=0: return immediately (fire-and-forget)
        timeout=300,
        # optional: route differently at trigger time
        work_queue_name="high-priority",
        job_variables={
            "env": {"CONFIG_KEY": "abc"},
            "image": "my-org/child-image:latest"
        },
    )

    # Inspect state and get the child flow's return value
    if child_run.state.is_completed():
        result = child_run.state.result()  # child flow's return value
        return {"child_result": result}
    else:
        raise RuntimeError(f"Child run ended in state {child_run.state.type}")

if __name__ == "__main__":
    parent_flow()

- Async parent flow (await if your flow is async):

from prefect import flow
from prefect.deployments import run_deployment

@flow
async def parent_flow_async():
    child_run = await run_deployment(
        name="child-flow/production",
        parameters={"param1": "value"},
        timeout=300
    )
    if child_run.state.is_completed():
        result = await child_run.state.result()
        return result

Notes and tips - Keeping the subflow link: It’s automatic when calling

run_deployment

from inside a flow (

as_subflow=True

by default). You’ll see the child nested under the parent in the UI. - Multiple deployments for the same child flow: Just choose which deployment to trigger by passing the corresponding

flow-name/deployment-name

. - Fire-and-forget: Use

timeout=0

. The child will still be linked as a subflow, but your parent won’t wait. - Overriding infra at trigger time: Use

work_queue_name

to route to a different queue or

job_variables

to override things like image, env, namespace, etc. Docs - How-to: Trigger ad-hoc deployment runs - API reference: prefect.deployments.flow_runs (run_deployment / arun_deployment) If you share: - the parent flow name - the child deployment name(s) - whether you want to wait or fire-and-forget …I can sketch the exact call you need.