src.integrations.kanban_client module#
Simple MCP Kanban Client for reliable task management.
This module provides a simplified client for interacting with the kanban-mcp server, focusing on reliability and following proven patterns that work consistently.
The client handles: - Task retrieval from kanban boards - Task assignment to agents - Board status monitoring - Automatic configuration loading
Notes
This implementation avoids persistent connections and creates a new MCP session for each operation to ensure reliability.
- class src.integrations.kanban_client.KanbanClient[source]#
Bases:
objectSimple MCP Kanban client that follows proven patterns for reliability.
This client creates a new MCP session for each operation rather than maintaining persistent connections, which has proven more reliable in practice.
Examples
>>> client = KanbanClient() >>> tasks = await client.get_available_tasks() >>> for task in tasks: ... print(f"Task: {task.name} - Priority: {task.priority.value}")
Notes
Planka credentials are loaded from environment variables or set to defaults. Board and project IDs are loaded from config_marcus.json if available.
- __init__()[source]#
Initialize the Simple MCP Kanban Client.
Loads configuration from config_marcus.json and sets up Planka environment variables. Config file takes precedence.
- Return type:
None
- property kanban_mcp_path: str#
Lazily resolve and cache the kanban-mcp executable path.
- Raises:
FileNotFoundError – If kanban-mcp cannot be found. Raised on first Planka call, not at construction time, so non-Planka providers are unaffected.
- async get_available_tasks()[source]#
Get all unassigned tasks from the kanban board.
Retrieves tasks that are in “available” states (TODO, BACKLOG, READY) and have not been assigned to any agent.
- Returns:
List of unassigned tasks sorted by priority
- Return type:
- Raises:
RuntimeError – If board_id is not set in configuration
Examples
>>> client = KanbanClient() >>> tasks = await client.get_available_tasks() >>> print(f"Found {len(tasks)} available tasks")
Notes
This method creates a new MCP session for the operation. Tasks are filtered based on their list name (TODO, BACKLOG, etc.) and whether they have an assigned_to field.
- async get_all_tasks()[source]#
Get all tasks from the kanban board regardless of status or assignment.
Retrieves tasks from all lists on the board, including assigned, unassigned, completed, and blocked tasks.
- Returns:
List of all tasks on the board
- Return type:
- Raises:
RuntimeError – If board_id is not set in configuration
Examples
>>> client = KanbanClient() >>> tasks = await client.get_all_tasks() >>> print(f"Total tasks on board: {len(tasks)}")
Notes
This method creates a new MCP session for the operation. Unlike get_available_tasks(), this includes tasks in all states and with any assignment status.
- async assign_task(task_id, agent_id)[source]#
Assign a task to an agent.
This method: 1. Adds a comment to the task indicating assignment 2. Moves the task to the “In Progress” list
- Parameters:
- Return type:
Examples
>>> await client.assign_task("card-123", "agent-001")
Notes
The task is automatically moved to the first list containing “progress” in its name (case-insensitive).
- async get_board_summary()[source]#
Get summary statistics for the kanban board.
- Returns:
Board statistics including task counts, completion percentage, and other metrics provided by the kanban-mcp server
- Return type:
- Raises:
RuntimeError – If board_id is not set in configuration
Examples
>>> summary = await client.get_board_summary() >>> print(f"Completion: {summary.get('completionPercentage', 0)}%")
Notes
The exact structure of the summary depends on the kanban-mcp implementation. Typically includes totalCards, completionPercentage, and counts by status.
- async add_comment(task_id, comment_text)[source]#
Add a comment to a task.
- Parameters:
- Return type:
Examples
>>> await client.add_comment("card-123", "Task completed successfully")
Notes
Comments are visible in the Planka UI and are timestamped automatically.
- async complete_task(task_id)[source]#
Mark a task as completed by moving it to the Done list.
Examples
>>> await client.complete_task("card-123")
Notes
The task is moved to the first list containing “done” or “completed” in its name (case-insensitive).
- async update_task_status(task_id, status)[source]#
Update a task’s status by moving it to the appropriate list.
- Parameters:
- Return type:
Examples
>>> await client.update_task_status("card-123", "blocked")
Notes
Status names are matched to list names containing the status keyword. For example, “blocked” matches any list with “blocked” in the name.
- async auto_setup_project(project_name, board_name='Main Board', project_root=None)[source]#
Automatically create a Planka project and board if they don’t exist.
This method will: 1. Create a new project in Planka 2. Create a new board in that project with default lists/labels 3. Save the IDs to .marcus_workspace.json 4. Load the IDs into memory
- Parameters:
- Returns:
Dictionary with project_id and board_id
- Return type:
Examples
>>> client = KanbanClient() >>> result = await client.auto_setup_project("My Project") >>> print(f"Project ID: {result['project_id']}") >>> print(f"Board ID: {result['board_id']}")
- async get_projects()[source]#
Get all projects from Planka.
- Returns:
List of projects with their details including: - id: Project ID - name: Project name - boards: List of boards in the project
- Return type:
Examples
>>> client = KanbanClient() >>> projects = await client.get_projects() >>> for project in projects: ... print(f"Project: {project['name']} (ID: {project['id']})")
Notes
This method creates a new MCP session for the operation. Useful for discovering existing projects in Planka.