Ravish Mallya
08/28/2025, 11:47 AMMarvin
08/28/2025, 11:47 AMMarvin
08/28/2025, 11:53 AMfrom 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:
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):
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:Marvin
08/28/2025, 11:53 AMfrom 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?