Skip to main content
Kanban enables agents to coordinate through persistent tasks, creating a shared workspace where work is tracked and distributed across multiple agents.
from praisonaiagents import Agent

coordinator = Agent(name="Coordinator", instructions="Break work into kanban tasks")
coordinator.start("Plan the auth feature and assign workers")
The user asks for a feature; the coordinator creates tasks workers pick up from the shared board.

Quick Start

1

Create Agent with Kanban Tools

from praisonaiagents import Agent
from praisonaiagents.kanban import KanbanStoreProtocol

# Agent with kanban protocols (implementation needed from wrapper)
agent = Agent(name="Coordinator", instructions="Break tasks down")
result = agent.start("Create user auth system")
2

Add Worker Agent

from praisonaiagents import Agent
from praisonaiagents.kanban import VALID_KANBAN_STATUSES

worker = Agent(name="Worker", instructions="Claim and complete tasks")
result = worker.start("Find ready tasks")

How It Works

Task coordination happens through a SQLite-backed persistent store that all agents and the UI share.
ComponentPurpose
Kanban StoreSQLite database storing tasks, comments, links
Agent Tools8 functions for task CRUD operations
CLI CommandsHuman interface for task management
Background DispatcherAuto-claims ready tasks for processing

Concepts

Task Statuses

Tasks flow through 8 defined states from creation to completion:

Boards

Boards provide workspace isolation for different projects or contexts: Tasks form directed acyclic graphs through parent-child dependencies:

Claim/Release

Workers coordinate through atomic claim and release operations:

Agent Tools

Kanban tools are available through the SDK protocols. The wrapper implementation provides these agent tools:

Task Management

ToolPurposeExample
kanban_createCreate new taskkanban_create("Implement auth", assignee="dev", idempotency_key="auth-v1")
kanban_listFilter taskskanban_list(status="ready", assignee="dev")
kanban_showGet task detailskanban_show("task_abc123")
kanban_runsRead attempt historykanban_runs("task_abc123")

Status Changes

ToolPurposeExample
kanban_completeMark task done with structured handoffkanban_complete("task_abc123", summary="Auth done; 14 tests pass", metadata={"changed_files": ["auth.py"]})
kanban_blockBlock with reasonkanban_block("task_abc123", "Need API keys")

Coordination

ToolPurposeExample
kanban_commentAdd progress notekanban_comment("task_abc123", "50% complete")
kanban_linkCreate dependencykanban_link("design_task", "implement_task")
kanban_heartbeatSignal livenesskanban_heartbeat("task_abc123", "testing")

Boards & Storage

Single Board (Default)

~/.praisonai/kanban.db

Multi-Board Layout

~/.praisonai/kanban/boards/
├── project-a/kanban.db
├── project-b/kanban.db
└── team-x/kanban.db

Environment Configuration

VariableEffectExample
PRAISONAI_KANBAN_BOARDSelect active boardexport PRAISONAI_KANBAN_BOARD=project-a
PRAISONAI_KANBAN_DBOverride DB pathexport PRAISONAI_KANBAN_DB=/custom/path.db

Attempt History & Retry

Each kanban task records every attempt, auto-blocks after repeated failures, and hands off a structured summary on completion.
1

Idempotent create with retry limit

from praisonaiagents import Agent
from praisonai.tools.kanban_tools import kanban_create

agent = Agent(name="Coordinator", instructions="Break work into tasks")

task = kanban_create(
    "Replace bcrypt with argon2",
    body="Migrate auth module to argon2",
    assignee="coder",
    max_retries=3,                       # per-task circuit-breaker
    idempotency_key="auth-refactor-q3",  # safe to repeat
)
2

Structured completion handoff

from praisonai.tools.kanban_tools import kanban_complete

kanban_complete(
    task["id"],
    summary="argon2 migration; 14 tests pass; CHANGELOG updated",
    metadata={
        "changed_files": ["auth.py", "tests/test_auth.py"],
        "tests_run": 14,
        "residual_risk": "rotate old hashes on next login",
    },
)
3

Read attempt history on retry

from praisonai.tools.kanban_tools import kanban_runs

history = kanban_runs(task["id"])
for run in history["runs"]:
    print(run["outcome"], run["error"] or run["summary"])
# crashed: ModuleNotFoundError: argon2
# completed: argon2 migration; 14 tests pass; CHANGELOG updated

What the dispatcher does automatically

When a worker claims a task, the dispatcher opens a run with the claim — anything the worker writes via kanban_complete closes it as completed, anything that crashes closes it as crashed/failed. Below the max_retries limit, failed attempts release the claim and put the task back on ready for re-dispatch. At the limit, the task auto-blocks with gave_up so a human can step in. A successful completion resets the failure counter — the limit is consecutive, not cumulative.

New params on kanban_create

OptionTypeDefaultDescription
max_retriesint3 (board default)Per-task circuit-breaker limit. After this many consecutive failed attempts the task auto-blocks. Invalid values (<1 / non-numeric) fall back to the board default.
idempotency_keystrNonePer-board, per-tenant dedup key. Repeating a create with the same key returns the existing task instead of a duplicate — safe for retrying automation and webhooks.

New params on kanban_complete

OptionTypeDefaultDescription
summarystr""Structured summary of what was done. Surfaced to linked children and retrying workers.
metadatadict{}Structured handoff fields. Common keys: changed_files, tests_run, residual_risk, next_steps.
commentstr""Free-text completion comment (kept for back-compat).
kanban_complete moves the task to done before recording the run. A move_task failure never leaves an orphaned completed run for a task that never reached done.

New tool: kanban_runs

ToolPurposeExample
kanban_runsRead attempt history (oldest first)kanban_runs("task_abc123"){"runs": [{"outcome": "crashed", "error": "...", "started_at": "...", "ended_at": "..."}, ...], "count": 2}
Each row matches the KanbanRunProtocol shape:
FieldTypeDescription
idintAuto-incremented run ID
task_idstrParent task ID
profilestrWorker/profile identifier
outcomestr | Nonecompleted, blocked, crashed, failed, or gave_up; None while the run is open
summarystrStructured handoff summary
metadatadictStructured handoff fields
errorstrError text for failed/crashed attempts
started_atstr | NoneISO-8601 timestamp
ended_atstr | NoneISO-8601 timestamp; None while the run is open

New Task fields

FieldTypeMeaning
max_retriesint | NonePer-task circuit-breaker limit. None means use the board default (3).
consecutive_failuresintCount of consecutive failed attempts since the last completed outcome. Reset to 0 by a successful completion.
current_run_idint | NoneActive attempt id; None when no run is open.
Existing SQLite databases are auto-migrated on first open — no user action required.

Common Patterns

Automation-safe creation — pass idempotency_key derived from the trigger (f"github-issue-{n}", f"webhook-{uuid}"). Re-firing the webhook is a no-op instead of spawning duplicates. Idempotency is per-board, per-tenant — two tenants on the same board never collide on the same key. Long-running flaky network tasks — set max_retries=5 per task to absorb transient flakes without giving up too early. Watch consecutive_failures in kanban_show to spot stuck tasks. Retrying workers use prior failures — call kanban_runs(task_id) at the start of a retry attempt to read prior outcomes and errors. Use summary and metadata from the most recent failed run as context: “previous attempt crashed at line X; avoid path Y.”
Webhooks, schedulers, and GitHub-issue triagers should always pass an idempotency_key. The cost is one extra string; the value is zero duplicate tasks under retry.
Fast deterministic work: max_retries=1. Flaky network or integration tests: max_retries=3–5. Leave the rest at the board default (3).
Use summary and metadata in kanban_complete instead of free-text comments. Downstream child tasks and retrying workers consume the structured fields; humans still read them fine.
The structured run history is the cheapest “what went wrong last time” context a retrying worker can fetch. Call kanban_runs(task_id) at the start of every retry attempt.

Common Patterns

Coordinator-Worker Pattern

# Coordinator breaks down requests
from praisonaiagents import Agent
from praisonaiagents.kanban import VALID_KANBAN_STATUSES

coordinator = Agent(
    name="Coordinator", 
    instructions="Break user requests into kanban tasks"
)

# Worker with heartbeat reporting
worker = Agent(
    name="Worker",
    instructions="Claim ready tasks, report progress via heartbeat"
)

Background Processing

# Background processing requires wrapper implementation
# The praisonaiagents.kanban protocols support:
# - Task claiming and status updates
# - Heartbeat reporting for long-running tasks
# - Multi-board coordination

Worker with Heartbeat

from praisonaiagents import Agent
from praisonaiagents.kanban import VALID_KANBAN_STATUSES

worker = Agent(
    name="Worker",
    instructions="""Claim ready tasks and report liveness via heartbeat. 
    Use kanban_heartbeat every 30 seconds during long-running work."""
)

# Worker claims task and reports progress
result = worker.start("Find ready tasks, claim one, and report progress")

Human-Agent Collaboration

# Human-agent collaboration pattern
# Requires wrapper implementation of:
# - CLI commands for task management
# - UI board visualization
# - Agent-to-store protocol bindings

Runs, Retries, and Structured Handoff

Kanban now tracks every execution attempt for each task via the task_runs table, supports per-task retry limits, and allows agents to pass structured data when completing a task.

kanban_complete — structured handoff

kanban_complete(
    task_id="task-123",
    summary="Implemented OAuth2 flow with PKCE",
    metadata={"pr_url": "https://github.com/org/repo/pull/42", "test_coverage": 0.91},
)
ParameterTypeDefaultDescription
task_idstr(required)Task to complete
summarystr""Human-readable completion summary stored in the run record
metadatadict{}Arbitrary structured data for downstream consumers (e.g. PR URL, metrics)

kanban_create — idempotency and retries

kanban_create(
    title="Write unit tests for auth module",
    body="Cover all edge cases in JWT validation",
    idempotency_key="auth-tests-v1",
    max_retries=3,
)
ParameterTypeDefaultDescription
titlestr(required)Task title
bodystr""Task description
idempotency_keystr | NoneNoneIf set, calling kanban_create again with the same key returns the existing task instead of creating a duplicate
max_retriesint0Max number of times the dispatcher retries a failed run. After N failures the task is auto-moved to blocked

kanban_runs — per-task run history

List all execution attempts for a task:
runs = kanban_runs(task_id="task-123")
for run in runs:
    print(run["outcome"], run["summary"], run["created_at"])
Each run record contains:
FieldTypeDescription
idstrUnique run ID
task_idstrParent task
outcomestr"success", "failure", or "in_progress"
summarystrSummary provided at completion
metadatadictStructured handoff data
created_atfloatUnix timestamp of run start
finished_atfloat | NoneUnix timestamp of run end

Auto-block after max retries

When max_retries > 0 and the task accumulates that many "failure" run outcomes, the dispatcher automatically moves the task from running to blocked. The task must be manually unblocked (kanban_move(task_id, "ready")) after the root cause is resolved.

Best Practices

Create tasks that can be completed in 15-30 minutes. Break larger work into linked subtasks using kanban_link for proper dependency tracking.
Move tasks through statuses systematically: todoreadyrunningdone. Use blocked for dependencies and review for human approval.
Use kanban_heartbeat during long-running tasks to signal liveness. Add detailed comments with kanban_comment to track progress and decisions.
Use separate boards for different projects or contexts. Default board works well for single-project setups.

Kanban CLI

CLI commands for viewing boards, tasks, and run history

Async Jobs

Asynchronous job processing and queuing

Background Tasks

Async job processing and scheduling

CLI Dispatcher

Command-line task orchestration