Dmytro Stelmakh
05/12/2026, 9:47 AMMarvin
05/12/2026, 9:48 AMMarvin
05/12/2026, 9:49 AMparent_flow_run_id.
In Prefect 3.x, a flow run is considered a subflow when it is created with a parent_task_run_id. Normally Prefect does this automatically when one flow calls another flow.
What that means in practice:
- Automatic path: parent_flow() calls child_flow() → Prefect creates the subflow relationship for you
- Manual path: if you're creating flow runs through the client API, you can pass parent_task_run_id when creating the child flow run
There is not a separate “mark this existing flow run as a subflow” operation after creation. The relationship has to be established when the child flow run is created.
Example APIs that support this:
- client.create_flow_run(..., parent_task_run_id=...)
- client.create_flow_run_from_deployment(..., parent_task_run_id=...)
So the answer is:
Yes, but only at flow-run creation time via `parent_task_run_id`; you can’t retroactively assign a parent/subflow relationship to an existing run.
If you just want normal subflow behavior, the recommended approach is still simply calling one flow from another.
If you want, I can also show:
1. the normal subflow pattern, or
2. a manual client/API example for creating a child flow run under a parent.Dmytro Stelmakh
05/12/2026, 10:21 AMMarvin
05/12/2026, 10:22 AMMarvin
05/12/2026, 10:24 AMself.result = self.flow_func(...)
and that call happens inside another running flow, then you're already getting subflow behavior in Prefect 3.x.
So in most cases, you do not need to switch anything:
- parameters are preserved automatically
- the parent/child relationship is created automatically
- Prefect sets the parent_task_run_id for you behind the scenes
So the recommended pattern is still just:
self.result = self.flow_func(**params)
or for positional args:
self.result = self.flow_func(arg1, arg2)
If self.flow_func is a @flow, and you're calling it from within another @flow, Prefect will register it as a child flow run.
Example:
from prefect import flow
class Runner:
def __init__(self, flow_func):
self.flow_func = flow_func
self.result = None
@flow
def parent(self, x, y):
self.result = self.flow_func(x=x, y=y)
return self.result
@flow
def child(x, y):
return x + y
runner = Runner(child)
runner.parent(1, 2)
That should already create:
- one parent flow run for parent
- one child/subflow run for child
When would you need to do something manual?
Only if you are not actually invoking the child flow from inside a running parent flow, and instead are creating flow runs via the client API.
In that case, you’d need to:
1. get the current parent flow run context
2. create a task run in the parent flow
3. create the child flow run with parent_task_run_id=<that task run id>
That looks more like this:
from prefect import flow
from prefect.context import FlowRunContext
from prefect.client.orchestration import get_client
from prefect.states import Pending
@flow
async def parent():
ctx = FlowRunContext.get()
if not ctx:
raise RuntimeError("Not running inside a flow")
async with get_client() as client:
parent_task_run = await client.create_task_run(
task={"name": "child-flow"},
flow_run_id=ctx.flow_run.id,
state=Pending(),
)
child_run = await client.create_flow_run(
flow=child,
parameters={"x": 1, "y": 2},
parent_task_run_id=parent_task_run.id,
state=Pending(),
)
return child_run.id
But that is advanced/manual orchestration, and usually not what you want.
Bottom line:
- If self.flow_func(...) is a flow call inside another flow: leave it as-is
- You do not need to manually preserve params or set parent_task_run_id
- Manual parent_task_run_id is only for client-created flow runs, not normal flow-to-flow calls
If you want, paste a small version of your class and I can tell you whether your current call is already creating a real subflow or whether you need the manual API route.