ckit_kanban

The ckit_kanban module provides functions for managing kanban board tasks. Bots use kanban boards to organize and track work.

Kanban Columns

Column	Purpose
`inbox`	New tasks arrive here (from messages, webhooks, etc.)
`todo`	Prioritized tasks ready to be worked on
`inprogress`	Currently being executed
`done`	Completed tasks

Task Flow

External Event          Bot Schedule           Scheduler
      │                      │                     │
      ▼                      ▼                     ▼
  ┌───────┐            ┌───────┐            ┌────────────┐
  │ INBOX │ ──────────►│  TODO │ ──────────►│ INPROGRESS │
  └───────┘  (sort)    └───────┘  (assign)  └────────────┘
                                                   │
                                                   │ (complete)
                                                   ▼
                                             ┌──────────┐
                                             │   DONE   │
                                             └──────────┘

Creating Tasks

bot_kanban_post_into_inbox()

Create a task in the inbox:

from flexus_client_kit import ckit_kanban

task = await ckit_kanban.bot_kanban_post_into_inbox(
    client=fclient,
    persona_id=rcx.persona.persona_id,
    title="Process new customer inquiry",
    description="Customer John asked about pricing",
    payload_json=json.dumps({
        "source": "slack",
        "channel": "#support",
        "customer_id": "123",
    }),
)
# Returns: FPersonaKanbanTaskOutput
print(f"Created task: {task.ktask_id}")

Parameters

Parameter	Type	Description
`client`	`FlexusClient`	API client
`persona_id`	`str`	Bot’s persona ID
`title`	`str`	Task title (shown in UI)
`description`	`str`	Detailed description
`payload_json`	`str`	JSON string with custom data
`fexp_name`	`str`	Expert name to handle task (default: “default”)

Moving Tasks

bot_kanban_move_task()

Move a task between columns:

# Move from inbox to todo (prioritize)
await ckit_kanban.bot_kanban_move_task(
    client=fclient,
    ktask_id=task.ktask_id,
    to_column="todo",
)

# Mark task as done
await ckit_kanban.bot_kanban_move_task(
    client=fclient,
    ktask_id=task.ktask_id,
    to_column="done",
    resolution="Completed successfully",
)

Parameters

Parameter	Type	Description
`client`	`FlexusClient`	API client
`ktask_id`	`str`	Task ID to move
`to_column`	`str`	Target column
`resolution`	`str`	Completion message (for “done”)

Scheduler Handles inprogress

You typically don’t move tasks to inprogress manually. The scheduler does this automatically when assigning work to the bot.

Querying Tasks

bot_kanban_list_tasks()

List tasks in a column:

tasks = await ckit_kanban.bot_kanban_list_tasks(
    client=fclient,
    persona_id=rcx.persona.persona_id,
    column="inbox",
    limit=50,
)

for task in tasks:
    print(f"{task.ktask_title}: {task.ktask_description}")

FPersonaKanbanTaskOutput

Task data structure:

Field	Type	Description
`ktask_id`	`str`	Unique task ID
`ktask_persona_id`	`str`	Owning persona
`ktask_column`	`str`	Current column
`ktask_title`	`str`	Task title
`ktask_description`	`str`	Description
`ktask_payload_json`	`str`	Custom JSON data
`ktask_fexp_name`	`str`	Expert to handle
`ktask_resolution`	`str`	Resolution message
`ktask_created_ts`	`datetime`	Creation time
`ktask_updated_ts`	`datetime`	Last update

Using in Tool via flexus_bot_kanban

The LLM can manage kanban through the built-in flexus_bot_kanban cloudtool. Include instructions in your system prompt:

SYSTEM_PROMPT = """
## Kanban Board

You have a kanban board to organize work. Use the `flexus_bot_kanban` tool:

- **post_inbox**: Create a new task in inbox
- **sort_inbox**: Move tasks from inbox to todo
- **resolve_task**: Mark current task as done

When you complete a task, always call `flexus_bot_kanban` with action="resolve_task".
"""

Event Handler for Tasks

React to task changes:

@rcx.on_updated_task
async def on_task_update(task: ckit_kanban.FPersonaKanbanTaskOutput):
    if task.ktask_column == "inprogress":
        # Task assigned to us
        payload = json.loads(task.ktask_payload_json or "{}")
        print(f"Starting work on: {task.ktask_title}")
        print(f"Payload: {payload}")

    elif task.ktask_column == "done":
        # Task completed
        print(f"Completed: {task.ktask_title}")
        print(f"Resolution: {task.ktask_resolution}")

Typical Patterns

Inbox Sorting (SCHED_TASK_SORT)

Bot runs periodically to prioritize inbox:

# In system prompt:
"""
When the inbox has tasks, review them and:
1. High priority tasks -> move to todo
2. Low priority tasks -> leave in inbox or resolve
3. Duplicates -> merge or resolve one

Use flexus_bot_kanban with action="sort_inbox" for each decision.
"""

Task Execution (SCHED_TODO)

Scheduler assigns tasks and starts conversations:

# Bot is activated with a task
# The task is automatically in "inprogress"
# System prompt should instruct to resolve when done:

"""
You have been assigned a task. Complete it and then call:
flexus_bot_kanban(action="resolve_task", resolution="<what you did>")
"""

Creating Tasks from Messages

@rcx.on_updated_message
async def on_message(msg):
    # Only process user messages from Slack integration
    if msg.ftm_role != "user":
        return

    provenance = json.loads(msg.ftm_provenance or "{}")
    if provenance.get("source") != "slack":
        return

    # Create kanban task for this message
    await ckit_kanban.bot_kanban_post_into_inbox(
        client=fclient,
        persona_id=rcx.persona.persona_id,
        title=f"Support request: {msg.ftm_content[:50]}...",
        description=msg.ftm_content,
        payload_json=json.dumps({
            "message_id": msg.ftm_id,
            "thread_id": msg.ftm_ft_id,
            "source": "slack",
            "channel": provenance.get("channel"),
        }),
    )

Complete Example

from flexus_client_kit import ckit_kanban
import json

async def handle_support_workflow(fclient, rcx):
    @rcx.on_updated_task
    async def on_task(task):
        if task.ktask_column != "inprogress":
            return

        # Parse task data
        payload = json.loads(task.ktask_payload_json or "{}")
        customer_id = payload.get("customer_id")

        # Do the work
        result = await process_support_request(
            customer_id=customer_id,
            description=task.ktask_description,
        )

        # Mark as done
        await ckit_kanban.bot_kanban_move_task(
            client=fclient,
            ktask_id=task.ktask_id,
            to_column="done",
            resolution=f"Resolved: {result}",
        )

    # Also listen for external events to create tasks
    slack = setup_slack_integration(fclient, rcx)
    slack.set_activity_callback(
        lambda activity, posted: ckit_kanban.bot_kanban_post_into_inbox(
            client=fclient,
            persona_id=rcx.persona.persona_id,
            title=f"Slack: {activity.text[:50]}",
            description=activity.text,
            payload_json=json.dumps({"slack_ts": activity.ts}),
        )
    )