<@ULVA73B9P> i have a parent flow that does a subp...
# ask-marvin
r
@Marvin i have a parent flow that does a subprocess call to a python script. the script internally triggers a flow run. i don’t have an option to make the script as a sub flow within the parent. Is there a way to build the subflow relationship between the parent and the flow subprocess script calls?
m
thought for 343.0 seconds
Short answer: You can’t “convert” an arbitrary flow run launched from another process into a subflow after the fact. However, you can create a true subflow relationship from a subprocess if you pass the parent task run ID into the subprocess and use it when creating the child flow run. Otherwise, use artifacts/tags to relate runs in the UI. Why: In Prefect 3.x, subflow linkage is recorded via the child run’s parent_task_run_id. That’s set at creation time. The UI’s parent/child tree is built from that field. What to do 1) Launch the subprocess from a task (so it has a task run id) - Wrap your subprocess call in a task, then pass the parent task run id into the script (via env var or CLI arg).
Copy code
from prefect import flow, task
from prefect.runtime import task_run
import os, subprocess

@task
def invoke_script():
    env = os.environ.copy()
    env["PREFECT_PARENT_TASK_RUN_ID"] = str(task_run.id())  # pass the parent task run id
    # also pass API config so the child can talk to your Prefect server/cloud
    # env["PREFECT_API_URL"] = "<https://api.prefect.cloud/api/accounts/.../workspaces/...>"
    # env["PREFECT_API_KEY"] = "<your key>"
    subprocess.run(["python", "child_launcher.py"], check=True, env=env)

@flow
def parent_flow():
    invoke_script()

if __name__ == "__main__":
    parent_flow()
2) In the subprocess script, create the child flow run with parent_task_run_id - Use the Prefect client to create the flow run (either from a deployment or directly), and include parent_task_run_id. Async example:
Copy code
import os, asyncio, uuid
from prefect.client.orchestration import get_client
from prefect.client.schemas.actions import FlowRunCreate

async def main():
    parent_task_run_id = os.environ.get("PREFECT_PARENT_TASK_RUN_ID")
    if not parent_task_run_id:
        raise RuntimeError("PREFECT_PARENT_TASK_RUN_ID was not provided to the script")

    async with get_client() as client:
        # Look up the deployment (replace with your "flow-name/deployment-name")
        dep = await client.read_deployment_by_name("my-flow/my-deployment")

        # Create the run and set the parent_task_run_id to link it as a subflow
        fr = await client.create_flow_run(
            FlowRunCreate(
                flow_id=dep.flow_id,
                deployment_id=dep.id,
                parent_task_run_id=uuid.UUID(parent_task_run_id),
                parameters={"x": 1},
                tags=["launched-from-subprocess"],
            )
        )
        print(f"Created child flow run: {fr.id}")

if __name__ == "__main__":
    asyncio.run(main())
Sync variant (if you prefer):
Copy code
import os, uuid
from prefect.client.orchestration import SyncPrefectClient
from prefect.client.schemas.actions import FlowRunCreate

with SyncPrefectClient() as client:
    parent_task_run_id = os.environ["PREFECT_PARENT_TASK_RUN_ID"]
    dep = client.read_deployment_by_name("my-flow/my-deployment")
    fr = client.create_flow_run(
        FlowRunCreate(
            flow_id=dep.flow_id,
            deployment_id=dep.id,
            parent_task_run_id=uuid.UUID(parent_task_run_id),
            parameters={"x": 1},
        )
    )
    print(f"Created child flow run: {fr.id}")
Notes and caveats - You cannot set “parent_flow_run_id” directly; Prefect determines a run’s parent flow run via the parent_task_run_id of the task that launched it. - If you call
prefect.deployments.run_deployment(...)
directly in the subprocess, there’s no active Prefect task context, so it will not establish a subflow relationship. Use the client pattern above to set parent_task_run_id. - If your parent currently spawns the script outside of a task (e.g., in flow body), wrap the subprocess call in a task so you have a task run id. - Make sure the subprocess has credentials to talk to your Prefect server/cloud (e.g., pass
PREFECT_API_URL
and
PREFECT_API_KEY
in the environment). If you can’t wire a true subflow - Use artifacts to cross-link runs in the UI:
- From the child run, create a link artifact pointing back to the parent’s UI URL. - Or write the child run’s UI URL as an artifact on the parent.
Copy code
from prefect.artifacts import create_link_artifact

create_link_artifact(
    key="parent-run",
    link="<https://app.prefect.cloud/.../flow-runs/flow-run/<PARENT_RUN_ID>>",
    description="Parent flow run"
)
- Add a shared tag (e.g.,
parent:<run_id>
) to both parent and child runs to make them easy to search together. Helpful references - run_deployment source: prefect.deployments.run_deployment - FlowRunCreate schema: FlowRunCreate - Artifacts API (source): prefect.artifacts Questions to tailor the solution - Are you launching the subprocess from within a task, or directly in the flow body? - How does the script currently “trigger a flow run” (Prefect client, run_deployment, REST call, something else)? - Can you modify the script to accept the parent task run id and create the flow run via the client as shown above?