Data Queries

These functions call the TaskFlow internal API using TASK_JWT.

list_tasks(status?, limit?, offset?) → list

List task instances in the project, newest first.

running = list_tasks(status="running")
for t in running:
    print(f"{t['name']} — {t['progress']}%")

Status values: pending, running, success, failed, pending_approval, rejected.

---

get_task(task_id) → dict

Non-blocking snapshot of a single task. Attaches result from the execution log for completed tasks.

task = get_task("550e8400-...")
if task["status"] == "success":
    print(task.get("result"))

---

list_agents() → list

List all agents in the project. Returns id, name, description, is_default, agent_type.

---

list_workspaces() → list

List all workspaces. Returns id, name, slug, type, is_default, artifact_type.

---

list_task_type_memories(task_type, query?, k, order, limit, offset) → list

Query memories for a task type. Two modes:

• Listing (no query): paginated by creation time
• Semantic search (query provided): similarity-ranked results

# Most relevant 5
hits = list_task_type_memories("summariser", query="auth flow", k=5)

# Least similar (most novel)
novel = list_task_type_memories("summariser", query="auth flow", k=5, order="least")

order: "most" (default) or "least". Semantic results include a score field.

---

get_task_memories(task_id, limit?, offset?) → list

Memories contributed by a single task run.

for m in get_task_memories(TASK_ID):
    print(m["memory"])

---

okf_search(query, workspace_id?) → list

Search the project knowledge base for OKF concepts.

hits = okf_search("authentication flow")
for h in hits:
    print(h["title"], "-", h.get("description", ""))

---

okf_lint(workspace_id?) → list

Run health checks on the knowledge base (broken links, orphans, duplicates, missing types).

issues = okf_lint()
for issue in issues:
    print(f"[{issue['issue_type']}] {issue['concept_id']}: {issue['message']}")