Yes — that pattern is fully supported. In Prefect 3.x: - You can call a plain Python function from inside a flow. - Inside that plain function, you can call any number of

@task

-decorated functions. - The plain function itself is not tracked by Prefect (it won’t show in the UI), but each

@task

call will create its own task run with state, logs, and timing in the UI. Key details: - Direct call vs submit: -

task_fn(args)

executes immediately and returns the actual result. -

task_fn.submit(args)

runs the task concurrently and returns a

PrefectFuture

you can wait on (e.g.,

f.result()

). - Dependencies are tracked automatically when passing results between tasks. - Retries, timeouts, caching, etc. apply at the task level. The undecorated helper has none of those unless you decorate it. - If you want the helper to appear in the UI or to attach retries/timeouts at that level, make it a

@task

(or a nested

@flow

if that’s conceptually a subflow). Example:

from prefect import flow, task

@task
def add(x, y):
    return x + y

@task
def double(z):
    return z * 2

def helper(a, b):
    # Each call below is a tracked task run
    s = add(a, b)          # returns result synchronously
    return double(s)

@flow
def my_flow(a: int, b: int):
    return helper(a, b)

if __name__ == "__main__":
    print(my_flow(3, 4))   # -> 14

Concurrent variant:

def helper_concurrent(values):
    futures = [add.submit(v, 1) for v in values]
    doubled = [double.submit(f) for f in futures]
    return [f.result() for f in doubled]

Quick check: Are you on Prefect 3.x, and do you want these inner tasks to run sequentially or concurrently? I can tailor an example either way.

Got it — thanks for clarifying you’re on Prefect 2. What you’re seeing in the UI is expected based on how Prefect 2 infers dependencies: - The UI graph shows data dependencies, not execution order. - Dependencies are inferred when a task consumes the output (future or result) of another task, or when you explicitly set

wait_for=[...]

. - If the first task’s output is passed (even indirectly, e.g., inside a dict or object) to downstream tasks, the UI will show that first task as an upstream of each consumer — a fan-out like you described. If you just ran tasks sequentially without passing outputs around, the graph won’t reflect order unless you add dependencies. To make “sequential” explicit in the graph, chain with

wait_for

(or pass futures in a chain): - Explicit sequential chain with wait_for:

from prefect import flow, task

@task
def t1(): ...
@task
def t2(): ...
@task
def t3(): ...

def helper():
    r1 = t1.submit()
    r2 = t2.submit(wait_for=[r1])
    r3 = t3.submit(wait_for=[r2])
    return r3.result()

@flow
def my_flow():
    return helper()

- Sequential via data dependencies (each consumes previous):

def helper():
    r1 = t1.submit()
    r2 = t2.submit(r1)  # r2 depends on r1
    r3 = t3.submit(r2)  # r3 depends on r2
    return r3.result()

- Purely sequential calls (no edges; UI won’t show order):

def helper():
    t1()  # runs first
    t2()  # runs second
    t3()  # runs third

Note: this runs sequentially due to how it’s written, but the UI will not show a chain unless you add dependencies as above. Common gotcha that causes the fan-out you’re seeing: - Accidentally reusing the first task’s result everywhere (e.g., building a shared context dict from

t1()

then passing that dict to all later tasks). That makes the first task a legitimate upstream of all consumers. If you can share a minimal snippet of your helper showing how you’re calling/passing values, I can point out exactly where the dependency inference is happening and suggest the smallest change to get the chain you want.

@Marvin I have a flow @flow run-pipeline() def run_pipeline: Data = init() Data = getwindow(Data) For window in Data[“windows”] Data = phase1(Data, window) Data = phase2(Data) Sendalert(Data) @task Def phase1(data): Data = taskone(data) Data = tasktwo(data) Def taskone(data): #code

Thanks — that helps. In Prefect 2, what you’re seeing happens when a single upstream value (or a dict that contains it) is passed into many downstream tasks. Prefect infers “data dependencies,” not just execution order, so the first task that produced

Data

will appear upstream of every task that consumes that same

Data

(directly or nested inside a dict). That’s why it looks like a fan‑out instead of a simple chain. Two fixes that keep the graph sequential and readable: 1) Keep orchestration in the flow (don’t decorate

phase1

), and pass only the latest future forward - Don’t keep reusing a dict that contains old futures - Chain with

.submit(...)

so edges reflect sequence

from prefect import flow, task

@task
def init(): ...
@task
def getwindow(data): ...
@task
def taskone(data, window): ...
@task
def tasktwo(data): ...
@task
def phase2(data): ...
@task
def sendalert(data): ...

def phase1(data_future, window):
    # sequential inside the helper via submitted futures
    t1 = taskone.submit(data_future, window)
    t2 = tasktwo.submit(t1)
    return t2  # return the future for the next step

@flow
def run_pipeline():
    data = init.submit()
    data = getwindow.submit(data)

    # Resolve just the windows list so it doesn't carry the upstream future
    windows = data.result()["windows"]

    for window in windows:
        data = phase1(data, window)   # passes the latest future only
        data = phase2.submit(data)

    sendalert.submit(data)

Result: a clean chain; no “first task upstream of everything” because only the current step’s future flows forward. 2) If you must pass a resolved Python value instead of futures, add explicit wait_for to encode order - Resolving values with

.result()

breaks data edges; add

wait_for

to retain sequencing in the UI

r1 = taskone.submit(data, window)
r2 = tasktwo.submit(r1)                 # data edge
r3 = phase2.submit(r2)                  # data edge
sendalert.submit(r3)                    # data edge

# or if you resolve to raw values somewhere:
val = r2.result()
r3 = phase2.submit(val, wait_for=[r2])  # explicit order edge

Common gotcha causing the fan‑out: - Carrying a single “Data” dict across steps that still contains an old future. When you pass that dict to later tasks, Prefect finds that future inside and links all those tasks back to the originating task. Fix by: - Rebuilding the dict from resolved values before passing it further, or - Returning just the minimal value needed for the next step (and avoid storing futures in long‑lived containers). If you can paste the real

phase1

and how

Data

is built/updated, I’ll point to the exact place a future is being reused and suggest the smallest change.