Task Management

create_task()

Create a new task instance from within a running script.

def create_task(
    identifier: str,
    name: str | None = None,
    params: dict | None = None,
    wait: bool = False,
    timeout: float = 300,
    poll_interval: float = 2,
    on_complete=None,
    on_error=None,
    on_feedback=None,
    show_in_kanban: bool | None = None,
    workspace_slug: str | None = None,
) -> str | dict

Parameter	Type	Default	Description
identifier	str	required	The task type's identifier field
name	str | None	auto-generated	Display name for the task instance
params	dict | None	{}	Parameters passed to the task
wait	bool	False	Block until complete; returns task dict with "result"
timeout	float	300	Timeout in seconds
poll_interval	float	2	Polling interval in seconds
on_complete	callable | None	None	Background callback (task_dict) -> None on success
on_error	callable | None	None	Error callback (exception) -> None
on_feedback	callable | None	None	Called when task enters pending_approval
show_in_kanban	bool | None	False for background services, True otherwise	Whether to show in kanban
workspace_slug	str | None	None	Override the task type's default workspace

Returns: UUID string (wait=False) or task dict with result (wait=True).

Example: blocking wait

def run(params: dict, reporter) -> None:
    reporter.set_phase("Running sub-task")
    result = create_task(
        "data-processor",
        params={"source": params["source"]},
        wait=True,
        timeout=120,
    )
    print(result["result"])

Example: fire-and-forget parallel

from workflow import create_task, flush_tasks

results = []

def run(params: dict, reporter) -> None:
    for url in params["urls"]:
        create_task(
            "fetch-url",
            params={"url": url},
            on_complete=lambda r: results.append(r["result"]),
            on_error=lambda e: print(f"Failed: {e}"),
            show_in_kanban=False,
        )
    flush_tasks()
    return results

---

wait_task()

Wait for a task to finish and return its result.

def wait_task(
    task_id: str,
    timeout: float = 300,
    poll_interval: float = 2,
    on_feedback=None,
) -> dict

Parameter	Type	Default	Description
task_id	str	required	UUID returned by create_task()
timeout	float	300	Seconds before RuntimeError is raised
poll_interval	float	2	Polling interval in seconds
on_feedback	callable | None	None	Called when task enters pending_approval; raises RuntimeError if not provided

Returns: Task dict with status, result, etc. Raises RuntimeError on failure.

---

feedback()

Send feedback to a sub-task in pending_approval state.

def feedback(task_id: str, message: str = "") -> None

• Agent tasks (with a Hermes session): message is sent as the next user turn; function blocks until the agent finishes that turn.
• Script tasks (no session): approves the task; message is ignored.

---

flush_tasks()

Wait for all background threads started via on_complete to finish.

def flush_tasks(timeout: float = 300) -> None

Call this at the end of run() when using on_complete callbacks.