Workflow Script Overview

A Workflow script is the core of TaskFlow task execution. Each task runs inside a Docker container that loads workflow.py on startup, then executes your task_script.py.

Two calling conventions

Function-based (recommended)

# Synchronous
def run(params: dict, reporter) -> None:
    reporter.set_phase("Processing")
    reporter.set_progress(50)

# Async (enables concurrent MCP calls)
async def run(params: dict, reporter) -> None:
    import asyncio
    r1, r2 = await asyncio.gather(
        async_mcp_call("exa", "search", {"query": "..."}),
        async_mcp_call("context7", "resolve", {"query": "..."}),
    )

Import-based (Agent / Hermes tasks)

from workflow import params, reporter, TASK_ID, project_status
# Script runs at module level — no run() function needed

Available symbols

Symbol	Description
params	Task parameter dict
reporter	Progress reporting
project_status	Reactive project-status proxy
project_tokens	Read-only project token dict
create_task	Spawn a sub-task
wait_task	Block until a task finishes
feedback	Send feedback to a pending-approval task
flush_tasks	Wait for background callbacks
mcp	Get an MCP client
mcp_call	One-liner sync MCP call
async_mcp_call	Async MCP call
list_tasks	Query task list
get_task	Query a single task
list_agents	Query agent list
list_workspaces	Query workspace list
list_task_type_memories	Query task-type memories
get_task_memories	Query memories for one task run
okf_search	Knowledge-base search
okf_lint	Knowledge-base health check
defineStore	Create a reactive state store