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?Bring your towel and join one of the fastest growing data communities. Welcome to our second-generation open source orchestration platform, a completely rethought approach to dataflow automation.
Powered by