src.integrations.providers package#

Kanban provider implementations.

class src.integrations.providers.Planka[source]#

Bases: KanbanInterface

Planka kanban board implementation using direct MCP client.

__init__(config)[source]#

Initialize Planka connection.

Parameters:

config (Dict[str, Any])

property board_id: str | None#

Get board ID from the client.

property project_id: str | None#

Get project ID from the client.

async connect()[source]#

Connect to Planka via MCP.

Return type:

bool

async disconnect()[source]#

Disconnect from Planka.

Return type:

None

async get_available_tasks()[source]#

Get unassigned tasks from backlog.

Return type:

List[Task]

async get_all_tasks()[source]#

Get all tasks from the board regardless of status or assignment.

Return type:

List[Task]

async get_task_by_id(task_id)[source]#

Get specific task by ID.

Return type:

Optional[Task]

Parameters:

task_id (str)

async create_task(task_data)[source]#

Create new task in Planka.

Return type:

Task

Parameters:

task_data (Dict[str, Any])

async update_task(task_id, updates)[source]#

Update task status or properties.

Return type:

Optional[Task]

Parameters:
async add_comment(task_id, comment)[source]#

Add comment to task.

Return type:

bool

Parameters:
async get_agent_tasks(agent_id)[source]#

Get all tasks assigned to a specific agent.

Return type:

List[Task]

Parameters:

agent_id (str)

async get_board_summary()[source]#

Get overall board statistics and summary.

Return type:

Dict[str, Any]

async assign_task(task_id, assignee_id)[source]#

Assign a task to a worker.

Return type:

bool

Parameters:

  • task_id (str)

  • assignee_id (str)

async move_task_to_column(task_id, column_name)[source]#

Move task to a specific column/status.

Return type:

bool

Parameters:

  • task_id (str)

  • column_name (str)

async get_project_metrics()[source]#

Get project metrics and statistics.

Return type:

Dict[str, Any]

async report_blocker(task_id, blocker_description, severity='medium')[source]#

Report a blocker on a task.

Return type:

bool

Parameters:

  • task_id (str)

  • blocker_description (str)

  • severity (str)

async update_task_progress(task_id, progress_data)[source]#

Update task progress.

Return type:

bool

Parameters:
async upload_attachment(task_id, filename, content, content_type=None)[source]#

Upload an attachment to a task.

Note: Attachment functionality not yet implemented for Planka integration. This is a placeholder to satisfy the abstract interface.

Parameters:
Return type:

Dict[str, Any]

Returns:

Dict with success=False indicating not implemented.

async get_attachments(task_id)[source]#

Get all attachments for a task.

Note: Attachment functionality not yet implemented for Planka integration. This is a placeholder to satisfy the abstract interface.

Parameters:

task_id (str)

Return type:

Dict[str, Any]

Returns:

Dict with empty attachments list.

async download_attachment(attachment_id, filename, task_id=None)[source]#

Download an attachment.

Note: Attachment functionality not yet implemented for Planka integration. This is a placeholder to satisfy the abstract interface.

Parameters:
Return type:

Dict[str, Any]

Returns:

Dict with success=False indicating not implemented.

class src.integrations.providers.PlankaKanban[source]#

Bases: KanbanInterface

Planka kanban board implementation.

__init__(config)[source]#

Initialize Planka connection.

Parameters:

config (Dict[str, Any]) –

Dictionary containing:

  • project_name: Name of the project in Planka

async connect()[source]#

Connect to Planka via MCP.

Returns:

True if connection successful, raises exception otherwise.

Return type:

bool

async disconnect()[source]#

Disconnect from Planka.

Return type:

None

async get_available_tasks()[source]#

Get unassigned tasks from backlog.

Returns:

List of available tasks from the backlog.

Return type:

List[Task]

async get_all_tasks()[source]#

Get all tasks from the board.

Returns:

List of all tasks on the board.

Return type:

List[Task]

async get_task_by_id(task_id)[source]#

Get specific task by ID.

Parameters:

task_id (str) – The ID of the task to retrieve.

Returns:

The task if found, None otherwise.

Return type:

Optional[Task]

async create_task(task_data)[source]#

Create new task in Planka.

Parameters:

task_data (Dict[str, Any]) – Dictionary containing task data (name, description, due_date).

Returns:

The newly created task object.

Return type:

Task

async update_task(task_id, updates)[source]#

Update existing task.

Parameters:

  • task_id (str) – The ID of the task to update.

  • updates (Dict[str, Any]) – Dictionary containing fields to update.

Returns:

The updated task object.

Return type:

Optional[Task]

async assign_task(task_id, assignee_id)[source]#

Assign task to worker.

Parameters:

  • task_id (str) – The ID of the task to assign.

  • assignee_id (str) – The ID of the worker to assign the task to.

Returns:

True if assignment successful.

Return type:

bool

async move_task_to_column(task_id, column_name)[source]#

Move task to specific column.

Parameters:

  • task_id (str) – The ID of the task to move.

  • column_name (str) – The name of the target column.

Returns:

True if move successful.

Return type:

bool

async add_comment(task_id, comment)[source]#

Add comment to task.

Parameters:

  • task_id (str) – The ID of the task to comment on.

  • comment (str) – The comment text to add.

Returns:

True if comment added successfully.

Return type:

bool

async get_project_metrics()[source]#

Get project metrics.

Returns:

Dictionary containing task counts by status.

Return type:

Dict[str, Any]

async report_blocker(task_id, blocker_description, severity='medium')[source]#

Report blocker on task.

Parameters:

  • task_id (str) – The ID of the task to report blocker for.

  • blocker_description (str) – Description of the blocker.

  • severity (str) – Severity level (default is “medium”).

Returns:

True if blocker reported successfully.

Return type:

bool

async update_task_progress(task_id, progress_data)[source]#

Update task progress.

Parameters:

  • task_id (str) – The ID of the task to update progress for.

  • progress_data (Dict[str, Any]) – Dictionary containing progress, status, and message.

Returns:

True if progress updated successfully.

Return type:

bool

async upload_attachment(task_id, filename, content, content_type=None)[source]#

Upload an attachment to a Planka card.

Uses the kanban-mcp attachment manager to upload files.

Parameters:

  • task_id (str) – The ID of the task to attach the file to.

  • filename (str) – Name of the file to upload.

  • content (Union[str, bytes]) – The file content (string or bytes).

  • content_type (Optional[str]) – MIME type of the content.

Returns:

Result dictionary with success status and attachment data.

Return type:

Dict[str, Any]

async get_attachments(task_id)[source]#

Get all attachments for a Planka card.

Parameters:

task_id (str) – The ID of the task to get attachments for.

Returns:

Result dictionary with success status and attachments list.

Return type:

Dict[str, Any]

async download_attachment(attachment_id, filename, task_id=None)[source]#

Download an attachment from Planka.

Parameters:

  • attachment_id (str) – The ID of the attachment to download.

  • filename (str) – The filename to use for the downloaded attachment.

  • task_id (Optional[str]) – The task ID (not used for Planka).

Returns:

Result dictionary with success status and attachment content.

Return type:

Dict[str, Any]

async delete_attachment(attachment_id, task_id=None)[source]#

Delete an attachment from Planka.

Parameters:

  • attachment_id (str) – The ID of the attachment to delete.

  • task_id (Optional[str]) – The task ID (not used for Planka).

Returns:

Result dictionary with success status.

Return type:

Dict[str, Any]

async update_attachment(attachment_id, filename=None, task_id=None)[source]#

Update attachment metadata in Planka.

Parameters:

  • attachment_id (str) – The ID of the attachment to update.

  • filename (Optional[str]) – New filename for the attachment.

  • task_id (Optional[str]) – The task ID (not used for Planka).

Returns:

Result dictionary with success status and updated data.

Return type:

Dict[str, Any]

class src.integrations.providers.LinearKanban[source]#

Bases: KanbanInterface

Linear kanban board implementation using MCP Server.

__init__(config)[source]#

Initialize Linear MCP connection.

Parameters:

config (Dict[str, Any]) –

Dictionary containing:

  • mcp_function_caller: Function to call MCP tools

  • team_id: Linear team ID

  • project_id: Optional Linear project ID

async connect()[source]#

Connect to Linear MCP Server.

Return type:

bool

async disconnect()[source]#

Disconnect from Linear MCP.

Return type:

None

async get_available_tasks()[source]#

Get unassigned tasks from backlog.

Return type:

List[Task]

async get_all_tasks()[source]#

Get all tasks from the board regardless of status or assignment.

Return type:

List[Task]

async get_task_by_id(task_id)[source]#

Get specific task by ID.

Return type:

Optional[Task]

Parameters:

task_id (str)

async create_task(task_data)[source]#

Create new task in Linear.

Return type:

Task

Parameters:

task_data (Dict[str, Any])

async update_task(task_id, updates)[source]#

Update existing task.

Return type:

Optional[Task]

Parameters:
async assign_task(task_id, assignee_id)[source]#

Assign task to user.

Return type:

bool

Parameters:

  • task_id (str)

  • assignee_id (str)

async move_task_to_column(task_id, column_name)[source]#

Move task to specific state.

Return type:

bool

Parameters:

  • task_id (str)

  • column_name (str)

async add_comment(task_id, comment)[source]#

Add comment to task.

Return type:

bool

Parameters:
async get_project_metrics()[source]#

Get project metrics.

Return type:

Dict[str, Any]

async report_blocker(task_id, blocker_description, severity='medium')[source]#

Report blocker on task.

Return type:

bool

Parameters:

  • task_id (str)

  • blocker_description (str)

  • severity (str)

async update_task_progress(task_id, progress_data)[source]#

Update task progress.

Return type:

bool

Parameters:
async upload_attachment(task_id, filename, content, content_type=None)[source]#

Upload an attachment to a task.

Linear MCP doesn’t currently support direct file uploads, so we’ll add the attachment content as a comment with metadata.

Return type:

Dict[str, Any]

Parameters:
async get_attachments(task_id)[source]#

Get all attachments for a task.

Since Linear MCP doesn’t have direct attachment support, this returns an empty list but maintains the interface.

Return type:

Dict[str, Any]

Parameters:

task_id (str)

async download_attachment(attachment_id, filename, task_id=None)[source]#

Download an attachment.

Linear MCP doesn’t support direct file attachments, so this returns an error message.

Return type:

Dict[str, Any]

Parameters:

  • attachment_id (str)

  • filename (str)

  • task_id (str | None)

class src.integrations.providers.GitHubKanban[source]#

Bases: KanbanInterface

GitHub Projects kanban board implementation using MCP Server.

__init__(config)[source]#

Initialize GitHub MCP connection.

Parameters:

config (Dict[str, Any]) –

Dictionary containing:

  • mcp_function_caller: Function to call MCP tools

  • owner: Repository owner (user or org)

  • repo: Repository name

  • project_number: Project number (for v2 projects)

async connect()[source]#

Connect to GitHub MCP Server.

Return type:

bool

async disconnect()[source]#

Disconnect from GitHub MCP.

Return type:

None

async get_available_tasks()[source]#

Get unassigned tasks from backlog.

Return type:

List[Task]

async get_task_by_id(task_id)[source]#

Get specific task by ID.

Return type:

Optional[Task]

Parameters:

task_id (str)

async create_task(task_data)[source]#

Create new issue in repository.

Return type:

Task

Parameters:

task_data (Dict[str, Any])

async update_task(task_id, updates)[source]#

Update existing issue.

Return type:

Optional[Task]

Parameters:
async assign_task(task_id, assignee_id)[source]#

Assign issue to user.

Return type:

bool

Parameters:

  • task_id (str)

  • assignee_id (str)

async move_task_to_column(task_id, column_name)[source]#

Move task to specific status column.

Return type:

bool

Parameters:

  • task_id (str)

  • column_name (str)

async add_comment(task_id, comment)[source]#

Add comment to issue.

Return type:

bool

Parameters:
async get_project_metrics()[source]#

Get project metrics.

Return type:

Dict[str, Any]

async report_blocker(task_id, blocker_description, severity='medium')[source]#

Report blocker on task.

Return type:

bool

Parameters:

  • task_id (str)

  • blocker_description (str)

  • severity (str)

async update_task_progress(task_id, progress_data)[source]#

Update task progress.

Return type:

bool

Parameters:
class src.integrations.providers.SQLiteKanban[source]#

Bases: KanbanInterface

SQLite-backed kanban board — zero external dependencies.

Parameters:

config (Dict[str, Any]) –

Provider configuration: - db_path : str

Path to SQLite database file. Default: ./data/kanban.db

  • project_namestr

    Default project name. Default: "Marcus Project"

  • attachments_dirstr

    Directory for attachment files. Default: ./data/attachments

__init__(config)[source]#

Initialize kanban provider with configuration.

Parameters:

config (Dict[str, Any]) – Provider-specific configuration - For Planka: url, username, password - For Linear: api_key, team_id - For GitHub: token, owner, repo, project_number

Return type:

None

async connect()[source]#

Create database and initialize schema.

Returns:

True if connection succeeded.

Return type:

bool

async disconnect()[source]#

Mark provider as disconnected.

Return type:

None

async auto_setup_project(project_name, board_name='Main Board', project_root=None)[source]#

Create a new project scope within this database.

Generates unique project_id and board_id so multiple experiments can share one SQLite file without task collisions.

Parameters:

  • project_name (str) – Name for the project.

  • board_name (str) – Name for the board (stored for reference).

  • project_root (Optional[str]) – Directory where agents write code.

Returns:

Dict with project_id and board_id.

Return type:

Dict[str, Any]

async create_task(task_data)[source]#

Create a new task on the board.

Parameters:

task_data (Dict[str, Any]) – Task fields including name, description, priority, labels, dependencies, estimated_hours, etc.

Returns:

The created task with generated ID and timestamps.

Return type:

Task

async get_available_tasks()[source]#

Get unassigned TODO tasks.

Returns:

Tasks with status=TODO and no assignment.

Return type:

List[Task]

async get_all_tasks()[source]#

Get all tasks regardless of status or assignment.

Returns:

All tasks on the board.

Return type:

List[Task]

async get_task_by_id(task_id)[source]#

Get a specific task by ID.

Parameters:

task_id (str) – The task’s unique identifier.

Returns:

The task if found, None otherwise.

Return type:

Optional[Task]

async update_task(task_id, updates)[source]#

Update task fields atomically.

Handles compound updates (e.g. status + assigned_to together).

Parameters:

  • task_id (str) – Task to update.

  • updates (Dict[str, Any]) – Fields to update. Supports: status, assigned_to, name, description, priority, completed_at, blocker, and any other Task column.

Returns:

Updated task, or None if task not found.

Return type:

Optional[Task]

async assign_task(task_id, assignee_id)[source]#

Assign task to an agent.

Sets assigned_to, moves to IN_PROGRESS, and adds a timestamped assignment comment.

Parameters:

  • task_id (str) – Task to assign.

  • assignee_id (str) – Agent ID to assign to.

Returns:

True if assignment succeeded.

Return type:

bool

async move_task_to_column(task_id, column_name)[source]#

Move task to a column by name.

Maps column name aliases to TaskStatus values (case-insensitive).

Parameters:

  • task_id (str) – Task to move.

  • column_name (str) – Target column name (e.g. “In Progress”, “done”, “backlog”, “on hold”).

Returns:

True if move succeeded.

Return type:

bool

async add_comment(task_id, comment)[source]#

Add a comment to a task.

Subtask IDs (formatted {parent_id}_sub_{N}) are automatically resolved to their parent task, since subtasks live in marcus.db’s subtasks collection rather than as rows in this DB’s tasks table. This matches how subtasks are presented to users (as checklist items on the parent card) and avoids FOREIGN KEY constraint failed errors that would otherwise drop the comment.

Parameters:

  • task_id (str) – Task to comment on. Subtask IDs are accepted and routed to the parent.

  • comment (str) – Comment content (supports markdown).

Returns:

True if the comment was stored. False if the resolved task does not exist or the insert failed for any other reason.

Return type:

bool

async report_blocker(task_id, blocker_description, severity='medium')[source]#

Report a blocker on a task.

Moves task to BLOCKED, stores blocker record, and adds a structured comment. Does NOT unassign the agent.

Parameters:

  • task_id (str) – Blocked task.

  • blocker_description (str) – Description of the impediment.

  • severity (str) – Blocker severity: “low”, “medium”, or “high”.

Returns:

True if blocker was recorded.

Return type:

bool

async update_task_progress(task_id, progress_data)[source]#

Update task progress with comment and conditional moves.

Parameters:

  • task_id (str) – Task to update.

  • progress_data (Dict[str, Any]) – Progress info with keys: - progress (int): 0-100 percentage - status (str): “in_progress”, “blocked”, “completed” - message (str): Progress description

Returns:

True if progress was recorded.

Return type:

bool

async get_project_metrics()[source]#

Get task counts grouped by status.

Returns:

Metrics dict with total_tasks, backlog_tasks, in_progress_tasks, completed_tasks, blocked_tasks.

Return type:

Dict[str, Any]

async create_project(name, description='')[source]#

Create a new project.

Parameters:

  • name (str) – Project name.

  • description (str) – Optional project description.

Returns:

Created project record with id, name, timestamps.

Return type:

Dict[str, Any]

async list_projects()[source]#

List all active projects.

Returns:

List of project records.

Return type:

List[Dict[str, Any]]

async delete_project(project_id, hard_delete=False)[source]#

Delete a project (soft delete by default).

Soft delete marks the project inactive. Hard delete removes the project and all its tasks, comments, blockers, labels, dependencies, and attachments.

Parameters:

  • project_id (str) – Project to delete.

  • hard_delete (bool) – If True, permanently remove all data. If False, mark as inactive (soft delete).

Returns:

True if project was found and deleted.

Return type:

bool

async get_project(project_id)[source]#

Get a single project by ID.

Parameters:

project_id (str) – Project ID.

Returns:

Project record, or None if not found.

Return type:

Optional[Dict[str, Any]]

async upload_attachment(task_id, filename, content, content_type=None)[source]#

Upload an attachment to a task.

Stores file on disk, metadata in SQLite.

Parameters:

  • task_id (str) – Task to attach to.

  • filename (str) – Name for the file.

  • content (Union[str, bytes]) – Base64-encoded string or raw bytes.

  • content_type (Optional[str]) – MIME type of the content.

Returns:

Result with success flag and attachment data.

Return type:

Dict[str, Any]

async get_attachments(task_id)[source]#

Get all attachments for a task.

Parameters:

task_id (str) – Task to get attachments for.

Returns:

Result with success flag and attachment list.

Return type:

Dict[str, Any]

async download_attachment(attachment_id, filename, task_id=None)[source]#

Download an attachment’s content.

Parameters:

  • attachment_id (str) – Attachment ID.

  • filename (str) – Expected filename.

  • task_id (Optional[str]) – Optional task ID for context.

Returns:

Result with base64-encoded content.

Return type:

Dict[str, Any]

async delete_attachment(attachment_id, task_id=None)[source]#

Delete an attachment (file + DB record).

Parameters:

  • attachment_id (str) – Attachment to delete.

  • task_id (Optional[str]) – Optional task ID for context.

Returns:

Result with success flag.

Return type:

Dict[str, Any]

Submodules#

On this page
Edit on GitHub
Show Source