windmill-labs/windmill · error · RuntimeError

task_flow("{path}") can only be called inside a @workflow

Error message

task_flow("{path}") can only be called inside a @workflow

What it means

Identical guard to task_script, but for flow steps: task_flow(path) returns a wrapper that must run inside a @workflow context so it can register the flow as the next step via ctx._next_step. Outside a workflow there is no step context to attach to, so the wrapper raises this RuntimeError.

Source

Thrown at python-client/wmill/wmill/client.py:3373

):
    """Create a task that dispatches to a separate Windmill flow.

    Usage::

        pipeline = task_flow("f/etl/pipeline", priority=10)

        @workflow
        async def main():
            result = await pipeline(input=data)
    """
    name = path.rsplit("/", 1)[-1]
    _opts = {k: v for k, v in {"timeout": timeout, "tag": tag, "cache_ttl": cache_ttl, "priority": priority, "concurrent_limit": concurrency_limit, "concurrency_key": concurrency_key, "concurrency_time_window_s": concurrency_time_window_s}.items() if v is not None} or None

    def wrapper(**kwargs):
        ctx = _workflow_ctx.get(None)
        if ctx is not None:
            return ctx._next_step(name, path, dispatch_type="flow", _task_options=_opts, **kwargs)
        raise RuntimeError(f'task_flow("{path}") can only be called inside a @workflow')

    wrapper.__name__ = name
    wrapper._is_task = True
    wrapper._task_path = path
    return wrapper


def workflow(func):
    """Decorator marking an async function as a workflow-as-code entry point.

    The function must be **deterministic**: given the same inputs it must call
    tasks in the same order on every replay. Branching on task results is fine
    (results are replayed from checkpoint), but branching on external state
    (current time, random values, external API calls) must use ``step()`` to
    checkpoint the value so replays see the same result.
    """
    func._is_workflow = True
    return func

View on GitHub (pinned to e474e8803c)

Solutions

1. Call it from inside a @wmill.workflow-decorated function executed as a flow
2. If you need to invoke a flow from a non-workflow script, use the run API (e.g. wmill client run flow endpoint / run_flow helper) instead of the task primitive
3. Guard dual-context helpers with _workflow_ctx.get(None) and fall back to a direct run when no ctx exists

Example fix

// before
result = task_flow("/u/flows/pipeline")(item)
// after
import wmill
@wmill.workflow
def main(item):
    return task_flow("/u/flows/pipeline")(item)

from wmill.client import _workflow_ctx
if _workflow_ctx.get(None) is None:
    raise RuntimeError("task_flow requires a @workflow context; use the run API instead")

def in_workflow() -> bool:
    from wmill.client import _workflow_ctx
    return _workflow_ctx.get(None) is not None

try:
    result = my_flow_task(item)
except RuntimeError as e:
    if "can only be called inside a @workflow" in str(e):
        result = launch_flow_via_api(item)
    else:
        raise

When it happens

Trigger: Invoking a wrapper created by wmill.task_flow(...) from a plain script, a standalone job, unit tests, or any code not executing under the @workflow decorator, causing _workflow_ctx.get() to be None.

Common situations: Reusing a helper module that wraps flows as tasks inside an ordinary script; testing a flow-wrapping function in pytest without a workflow runtime; migrating code from a flow to a script while keeping the task_flow wiring.

Related errors

• task_script("{path}") can only be called inside a @workflow
• wait_for_approval can only be called inside a @workflow
• could not read the run of job ${id}: ${res.status} ${await r
• could not record run ${experiment_id}: ${res.status} ${await
• file path escapes the build directory: ${rel}

AI-assisted analysis of windmill-labs/windmill@e474e8803c (2026-09-03). Data as JSON: /api/errors/de7245da438e5125. Report an issue: GitHub.