{"record":{"id":"de7245da438e5125","repo":"windmill-labs/windmill","slug":"task-flow-path-can-only-be-called-inside-a-w","errorCode":null,"errorMessage":"task_flow(\"{path}\") can only be called inside a @workflow","messagePattern":"task_flow\\(\"(.+?)\"\\) can only be called inside a @workflow","errorType":"exception","errorClass":"RuntimeError","httpStatus":null,"severity":"error","filePath":"python-client/wmill/wmill/client.py","lineNumber":3373,"sourceCode":"):\n    \"\"\"Create a task that dispatches to a separate Windmill flow.\n\n    Usage::\n\n        pipeline = task_flow(\"f/etl/pipeline\", priority=10)\n\n        @workflow\n        async def main():\n            result = await pipeline(input=data)\n    \"\"\"\n    name = path.rsplit(\"/\", 1)[-1]\n    _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\n\n    def wrapper(**kwargs):\n        ctx = _workflow_ctx.get(None)\n        if ctx is not None:\n            return ctx._next_step(name, path, dispatch_type=\"flow\", _task_options=_opts, **kwargs)\n        raise RuntimeError(f'task_flow(\"{path}\") can only be called inside a @workflow')\n\n    wrapper.__name__ = name\n    wrapper._is_task = True\n    wrapper._task_path = path\n    return wrapper\n\n\ndef workflow(func):\n    \"\"\"Decorator marking an async function as a workflow-as-code entry point.\n\n    The function must be **deterministic**: given the same inputs it must call\n    tasks in the same order on every replay. Branching on task results is fine\n    (results are replayed from checkpoint), but branching on external state\n    (current time, random values, external API calls) must use ``step()`` to\n    checkpoint the value so replays see the same result.\n    \"\"\"\n    func._is_workflow = True\n    return func","sourceCodeStart":3355,"sourceCodeEnd":3391,"githubUrl":"https://github.com/windmill-labs/windmill/blob/e474e8803ce2ff5c2df09a58dab51d45f5c922ca/python-client/wmill/wmill/client.py#L3355-L3391","documentation":"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.","triggerScenarios":"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.","commonSituations":"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.","solutions":["Call it from inside a @wmill.workflow-decorated function executed as a flow","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","Guard dual-context helpers with _workflow_ctx.get(None) and fall back to a direct run when no ctx exists"],"exampleFix":"// before\nresult = task_flow(\"/u/flows/pipeline\")(item)\n// after\nimport wmill\n@wmill.workflow\ndef main(item):\n    return task_flow(\"/u/flows/pipeline\")(item)","handlingStrategy":"validation","validationCode":"from wmill.client import _workflow_ctx\nif _workflow_ctx.get(None) is None:\n    raise RuntimeError(\"task_flow requires a @workflow context; use the run API instead\")","typeGuard":"def in_workflow() -> bool:\n    from wmill.client import _workflow_ctx\n    return _workflow_ctx.get(None) is not None","tryCatchPattern":"try:\n    result = my_flow_task(item)\nexcept RuntimeError as e:\n    if \"can only be called inside a @workflow\" in str(e):\n        result = launch_flow_via_api(item)\n    else:\n        raise","preventionTips":["Call task_flow wrappers only within @wmill.workflow functions","Never call @task wrappers at module import time","Keep flow-orchestration code out of plain scripts and unit tests; mock it instead"],"tags":["python","windmill","workflow-context","misuse"],"backgroundTag":"workflow-context-required","analyzedSha":"e474e8803ce2ff5c2df09a58dab51d45f5c922ca","analyzedAt":"2026-09-03T12:38:19.024Z","contentChangedAt":"2026-09-03T12:38:19.024Z","schemaVersion":2},"datasetVersion":"2026-09-08T10:18:20.063Z"}