dachdigitalwerksorg/kanvaflow
KanvaFlow — Remixable team task board with RBAC, drag-and-drop Kanban, dependencies, Dodil Models AI, and MCP tools.
A remix template · 0 remixes · Remix dachdigitalwerksorg/kanvaflow
app/boards/doors/mcp.py
"""The MCP Door (`/mcp`): Claude and ChatGPT action tools backed by the exact same use cases.
Every tool resolves the caller from the Dodil gateway's `X-Dodil-User` header, so an AI agent
only sees and modifies boards its signed-in human is permitted to access.
Tools expose high-level domain actions — never raw SQL or table access.
"""
from typing import Optional
from mcp.server.mcpserver import Context, MCPServer
from mcp.server.mcpserver.exceptions import ToolError
from boards.application.use_cases import Boards, NotFound
from boards.domain.model import Forbidden, Invalid
from boards.doors.identity import person_from
INSTRUCTIONS = (
"KanvaFlow team task boards. Use `my_boards` or `my_agenda` first to orient yourself. "
"Boards accept either their `board_id` or exact board title (e.g. 'Ops'). "
"Default lists are 'To do', 'Doing', and 'Done' (custom lists are also supported). "
"Use `list_overdue`, `what_is_blocking`, `search_similar_cards`, `suggest_checklist`, "
"and `summarize_board` to help the user plan and unblock work."
)
def build_mcp(boards: Boards) -> MCPServer:
server = MCPServer(name="kanvaflow", instructions=INSTRUCTIONS)
def call(ctx: Context, fn, *args, **kwargs):
person = person_from(ctx.headers or {})
if person is None:
raise ToolError("401: sign in first")
try:
return fn(person, *args, **kwargs)
except (NotFound, Forbidden, Invalid) as exc:
raise ToolError(str(exc)) from exc
@server.tool()
def my_boards(ctx: Context) -> list:
"""List all boards you belong to, including your role, card count, and overdue count."""
return call(ctx, boards.my_boards)
@server.tool()
def show_board(board_id: str, ctx: Context) -> dict:
"""Show a board (by board_id or exact board title), its lists, members, and all cards."""
return call(ctx, boards.show_board, board_id)
@server.tool()
def create_board(title: str, ctx: Context, description: str = "") -> dict:
"""Create a new board with 'To do', 'Doing', and 'Done' lists. You become its owner."""
return call(ctx, boards.create_board, title, description)
@server.tool()
def add_member(board_id: str, email: str, ctx: Context, role: str = "member") -> dict:
"""Owner only: invite someone to a board with role 'member' (can edit) or 'viewer' (read-only)."""
return call(ctx, boards.add_member, board_id, email, role)
@server.tool()
def remove_member(board_id: str, email: str, ctx: Context) -> dict:
"""Owner only: remove a member or viewer from a board."""
return call(ctx, boards.remove_member, board_id, email)
@server.tool()
def add_list(board_id: str, name: str, ctx: Context, is_done: bool = False) -> dict:
"""Add a custom workflow list column to a board (set is_done=True if cards in it are completed)."""
return call(ctx, boards.add_list, board_id, name, is_done)
@server.tool()
def add_card(
board_id: str,
title: str,
ctx: Context,
list_name: str = "To do",
description: str = "",
assignee: Optional[str] = None,
due_date: Optional[str] = None,
labels: Optional[list[str]] = None,
) -> dict:
"""Add a card to a board (accepts board_id or exact board title). due_date format: YYYY-MM-DD."""
return call(
ctx,
boards.add_card,
board_id,
title,
list_name=list_name,
description=description,
assignee=assignee,
due_date=due_date,
labels=labels,
)
@server.tool()
def update_card(
card_id: str,
ctx: Context,
title: Optional[str] = None,
description: Optional[str] = None,
assignee: Optional[str] = None,
due_date: Optional[str] = None,
labels: Optional[list[str]] = None,
) -> dict:
"""Update a card's title, description, assignee, due_date (YYYY-MM-DD, or '' to clear), or labels."""
kwargs = {}
if title is not None:
kwargs["title"] = title
if description is not None:
kwargs["description"] = description
if assignee is not None:
kwargs["assignee"] = assignee
if due_date is not None:
kwargs["due_date"] = due_date
if labels is not None:
kwargs["labels"] = labels
return call(ctx, boards.update_card, card_id, **kwargs)
@server.tool()
def assign_card(card_id: str, assignee: str, ctx: Context) -> dict:
"""Assign a card to a team member's email address (pass '' to unassign)."""
return call(ctx, boards.assign_card, card_id, assignee)
@server.tool()
def move_card(card_id: str, list_name: str, ctx: Context, position: Optional[int] = None) -> dict:
"""Move a card to another list (e.g. 'To do', 'Doing', 'Done') and optional position."""
return call(ctx, boards.move_card, card_id, list_name, position=position)
@server.tool()
def delete_card(card_id: str, ctx: Context) -> dict:
"""Delete a card from its board."""
return call(ctx, boards.delete_card, card_id)
@server.tool()
def my_agenda(ctx: Context, days: int = 7) -> dict:
"""Answer 'what's on my plate this week?': your overdue, due-this-week, and in-progress cards."""
return call(ctx, boards.my_agenda, days=days)
@server.tool()
def list_overdue(ctx: Context, board_id: Optional[str] = None) -> list:
"""Answer 'what's overdue?': list all unfinished overdue cards across one board or all your boards."""
return call(ctx, boards.list_overdue, board_id=board_id)
@server.tool()
def add_blocker(card_id: str, blocked_by_id: str, ctx: Context) -> dict:
"""Mark `card_id` as blocked by `blocked_by_id` on the same board (rejects circular dependencies)."""
return call(ctx, boards.add_blocker, card_id, blocked_by_id)
@server.tool()
def remove_blocker(card_id: str, blocked_by_id: str, ctx: Context) -> dict:
"""Remove a 'blocked by' dependency link from `card_id`."""
return call(ctx, boards.remove_blocker, card_id, blocked_by_id)
@server.tool()
def what_is_blocking(target: str, ctx: Context, board_id: Optional[str] = None) -> dict:
"""Answer 'what is blocking the launch?': inspect unfinished direct & upstream blockers by card ID or title keyword."""
return call(ctx, boards.what_is_blocking, target, board_id=board_id)
@server.tool()
def board_activity(board_id: str, ctx: Context, limit: int = 30) -> list:
"""List recent audit activity on a board (who, when, what changed)."""
return call(ctx, boards.board_activity, board_id, limit=limit)
@server.tool()
def card_history(card_id: str, ctx: Context) -> list:
"""List full change history for a specific card."""
return call(ctx, boards.card_history, card_id)
@server.tool()
def suggest_checklist(card_id: str, ctx: Context, apply: bool = False) -> dict:
"""AI (Dodil Models): turn a card into a 3-6 step checklist. Set apply=True to save steps onto the card."""
return call(ctx, boards.suggest_checklist, card_id, apply=apply)
@server.tool()
def search_similar_cards(
query: str,
ctx: Context,
board_id: Optional[str] = None,
limit: int = 5,
) -> list:
"""AI (Dodil Models): find cards similar in meaning to a query or card_id across visible boards."""
return call(ctx, boards.search_similar_cards, query, board_id=board_id, limit=limit)
@server.tool()
def summarize_board(board_id: str, ctx: Context) -> dict:
"""AI (Dodil Models): write a short daily standup summary of progress, overdue work, and blockers."""
return call(ctx, boards.summarize_board, board_id)
return server