{"record":{"id":"476f9cf5a95d9643","repo":"windmill-labs/windmill","slug":"task-script-path-can-only-be-called-inside-a","errorCode":null,"errorMessage":"task_script(\"{path}\") can only be called inside a @workflow","messagePattern":"task_script\\(\"(.+?)\"\\) can only be called inside a @workflow","errorType":"exception","errorClass":"RuntimeError","httpStatus":null,"severity":"error","filePath":"python-client/wmill/wmill/client.py","lineNumber":3337,"sourceCode":"):\n    \"\"\"Create a task that dispatches to a separate Windmill script.\n\n    Usage::\n\n        extract = task_script(\"f/data/extract\", timeout=600)\n\n        @workflow\n        async def main():\n            data = await extract(url=\"https://...\")\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=\"script\", _task_options=_opts, **kwargs)\n        raise RuntimeError(f'task_script(\"{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 task_flow(\n    path: str,\n    *,\n    timeout: Optional[int] = None,\n    tag: Optional[str] = None,\n    cache_ttl: Optional[int] = None,\n    priority: Optional[int] = None,\n    concurrency_limit: Optional[int] = None,\n    concurrency_key: Optional[str] = None,\n    concurrency_time_window_s: Optional[int] = None,\n):","sourceCodeStart":3319,"sourceCodeEnd":3355,"githubUrl":"https://github.com/windmill-labs/windmill/blob/e474e8803ce2ff5c2df09a58dab51d45f5c922ca/python-client/wmill/wmill/client.py#L3319-L3355","documentation":"task_script(path) returns a wrapper that, when invoked, enqueues a Windmill script as a step — but only if a @workflow-decorated context exists in the current thread-local _workflow_ctx. When called outside a workflow, the wrapper deliberately raises this RuntimeError instead of silently running the script as a plain sub-call. It is a guard against using workflow-only task primitives in ordinary scripts.","triggerScenarios":"Calling a function produced by wmill.task_script(...) inside a plain Python script, a Job/standalone run, a worker context, or any code path that is not executed under the @workflow (wmill.workflow) decorator — i.e. _workflow_ctx.get() returned None.","commonSituations":"Moving task-heavy code into a plain script for testing without realizing task_script only resolves within a workflow; importing a shared module that calls a @task at module level; refactoring a flow step into a standalone script and forgetting that step wiring (ctx._next_step) requires the workflow runtime.","solutions":["Wrap the calling code in @wmill.workflow and run it as a flow so _workflow_ctx is populated","Replace the task_script call with a direct sub-run API (e.g. wmill.run_script or resume APIs) if you actually want to launch a script outside a flow","If the code must work in both contexts, check ctx first: ctx = _workflow_ctx.get(None); branch to direct execution when ctx is None"],"exampleFix":"// before\ndef process():\n    return task_script(\"/u/scripts/transform\")(payload)\n// after\nimport wmill\n@wmill.workflow\ndef process():\n    return task_script(\"/u/scripts/transform\")(payload)","handlingStrategy":"validation","validationCode":"import wmill\nfrom wmill.client import _workflow_ctx\nif _workflow_ctx.get(None) is None:\n    raise RuntimeError(\"task_script requires a @workflow context\")\n# or: only call tasks inside @wmill.workflow functions","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_task(x)\nexcept RuntimeError as e:\n    if \"can only be called inside a @workflow\" in str(e):\n        result = run_directly(x)  # fallback outside flows\n    else:\n        raise","preventionTips":["Only invoke @task-decorated wrappers inside @wmill.workflow functions","Add an in_workflow() assert in shared helpers that use task primitives","When testing tasks locally, replace them with direct function calls or the run API"],"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"}