dmslanes/lanes

Lanes - team task boards with roles, history and an MCP endpoint for Claude and ChatGPT

A remix template · 0 remixes · Remix dmslanes/lanes

app/mcp/tools.py

"""MCP door: tools are actions (move_card, assign_card), never table access.

Each tool reads the signed-in person from the gateway headers and calls the same
use case as the web route, so an assistant sees exactly the boards its person sees.
"""
from __future__ import annotations

from datetime import date

from mcp.server.mcpserver import Context, MCPServer
from mcp.server.mcpserver.exceptions import ToolError
from mcp.types import ToolAnnotations

from app.application.use_cases import BoardService
from app.application.views import activity_view, board_view, cards_with_context, full_board_view, member_view
from app.domain.model import DomainError, NotFound
from app.infrastructure.identity import Unauthenticated, email_from_headers

READ = ToolAnnotations(read_only_hint=True, destructive_hint=False, idempotent_hint=True, open_world_hint=False)
WRITE = ToolAnnotations(read_only_hint=False, destructive_hint=False, idempotent_hint=False, open_world_hint=False)
DELETE = ToolAnnotations(read_only_hint=False, destructive_hint=True, idempotent_hint=True, open_world_hint=False)

INSTRUCTIONS = (
    "Lanes is a team task board: workspaces hold boards, boards hold lists (To do, Doing, Done...), lists hold "
    "cards. Cards are referred to by number (\"card 12\"). Boards and lists can be named instead of using ids "
    "(\"the Ops board\", \"Doing\"). You only see the boards the signed-in person is a member of; viewers can "
    "read but not change anything."
)


def build_mcp(svc: BoardService) -> MCPServer:
    server = MCPServer(name="lanes", instructions=INSTRUCTIONS)

    def actor(ctx: Context) -> str:
        try:
            return email_from_headers(ctx.headers or {})
        except Unauthenticated as e:
            raise ToolError(f"401: {e}") from None

    def run(fn):
        try:
            return fn()
        except DomainError as e:
            raise ToolError(f"{e.code}: {e}") from None

    def board_id(who: str, board: str) -> str:
        return svc.find_board(who, board).id

    def workspace_id(who: str, workspace: str | None) -> str:
        spaces = svc.list_workspaces(who)
        if workspace:
            for ws, _ in spaces:
                if workspace in (ws.id, ws.name) or ws.name.lower() == workspace.lower():
                    return ws.id
            raise NotFound(f"no workspace {workspace!r} that you belong to")
        if len(spaces) == 1:
            return spaces[0][0].id
        if not spaces:
            raise NotFound("you are not in any workspace yet; create one first")
        raise DomainError("you are in several workspaces; say which: " + ", ".join(ws.name for ws, _ in spaces))

    @server.tool(annotations=READ)
    def list_boards(ctx: Context) -> list[dict]:
        """Every board the signed-in person can see, with its workspace and their role."""
        who = actor(ctx)
        return run(lambda: [board_view(b) for b in svc.list_boards(who)])

    @server.tool(annotations=READ)
    def show_board(board: str, ctx: Context) -> dict:
        """A board's lists and their cards. `board` is a name ("Ops") or an id."""
        who = actor(ctx)
        return run(lambda: full_board_view(svc.get_board(who, board_id(who, board))))

    @server.tool(annotations=READ)
    def my_cards(ctx: Context, days: int | None = None) -> list[dict]:
        """Open cards assigned to me ("what's on my plate"). `days`=7 limits to cards due within a week,
        overdue included; leave it out for every open card of mine."""
        who = actor(ctx)
        return run(lambda: cards_with_context(svc, svc.my_cards(who, days)))

    @server.tool(annotations=READ)
    def overdue_cards(ctx: Context, board: str | None = None) -> list[dict]:
        """Cards past their due date that are not in a Done list, across my boards or on one board."""
        who = actor(ctx)
        return run(lambda: cards_with_context(svc, svc.overdue(who, board_id(who, board) if board else None)))

    @server.tool(annotations=READ)
    def search_cards(text: str, ctx: Context) -> list[dict]:
        """Cards whose title, description or labels contain every word of `text`."""
        who = actor(ctx)
        return run(lambda: cards_with_context(svc, svc.search_cards(who, text)))

    @server.tool(annotations=READ)
    def card_history(card: str, ctx: Context) -> list[dict]:
        """Who changed card N, when, and what changed."""
        who = actor(ctx)
        return run(lambda: [activity_view(a) for a in svc.card_history(who, card)])

    @server.tool(annotations=WRITE)
    def add_card(board: str, list: str, title: str, ctx: Context, description: str = "",
                 assignee: str | None = None, due_date: date | None = None,
                 labels: list[str] | None = None) -> dict:
        """Add a card to a list of a board, e.g. board="Ops", list="Doing", title="Send invoice".
        `assignee` is a member's email; `due_date` is YYYY-MM-DD."""
        who = actor(ctx)
        return run(lambda: cards_with_context(svc, [svc.add_card(
            who, board_id(who, board), list, title, description, assignee, due_date, labels)])[0])

    @server.tool(annotations=WRITE)
    def move_card(card: str, to_list: str, ctx: Context, position: int | None = None) -> dict:
        """Move card N to another list of its board ("move card 12 to Done"). `position` 0 = top."""
        who = actor(ctx)
        return run(lambda: cards_with_context(svc, [svc.move_card(who, card, to_list, position)])[0])

    @server.tool(annotations=WRITE)
    def assign_card(card: str, ctx: Context, assignee: str | None = None) -> dict:
        """Give card N to a workspace member (email), or leave `assignee` empty to unassign."""
        who = actor(ctx)
        return run(lambda: cards_with_context(svc, [svc.assign_card(who, card, assignee)])[0])

    @server.tool(annotations=WRITE)
    def update_card(card: str, ctx: Context, title: str | None = None, description: str | None = None,
                    due_date: date | None = None, labels: list[str] | None = None) -> dict:
        """Change card N's title, description, due date (YYYY-MM-DD) or labels. Fields left out stay as they are."""
        who = actor(ctx)
        fields = {k: v for k, v in dict(title=title, description=description, due_date=due_date,
                                         labels=labels).items() if v is not None}
        return run(lambda: cards_with_context(svc, [svc.update_card(who, card, **fields)])[0])

    @server.tool(annotations=DELETE)
    def delete_card(card: str, ctx: Context) -> str:
        """Delete card N (its history keeps a 'deleted' entry)."""
        who = actor(ctx)
        run(lambda: svc.delete_card(who, card))
        return f"card {card} deleted"

    @server.tool(annotations=WRITE)
    def block_card(card: str, blocked_by: str, ctx: Context) -> dict:
        """Card N can't finish before card M: "card 4 is blocked by card 2". Loops are refused."""
        who = actor(ctx)
        return run(lambda: cards_with_context(svc, [svc.block_card(who, card, blocked_by)])[0])

    @server.tool(annotations=WRITE)
    def unblock_card(card: str, blocked_by: str, ctx: Context) -> dict:
        """Remove the "blocked by" link between card N and card M."""
        who = actor(ctx)
        return run(lambda: cards_with_context(svc, [svc.unblock_card(who, card, blocked_by)])[0])

    @server.tool(annotations=READ)
    def what_is_blocking(card: str, ctx: Context) -> list[dict]:
        """Open cards standing between a card and Done, at any depth ("what is blocking the launch?").
        `card` is a number or words from its title."""
        who = actor(ctx)
        return run(lambda: cards_with_context(svc, svc.blockers(who, card)))

    @server.tool(annotations=WRITE)
    def card_to_checklist(card: str, ctx: Context) -> dict:
        """AI (Dodil Models): turn card N into a checklist of concrete steps, appended to its description."""
        who = actor(ctx)
        return run(lambda: cards_with_context(svc, [svc.card_to_checklist(who, card)])[0])

    @server.tool(annotations=READ)
    def find_similar_cards(card: str, ctx: Context, limit: int = 5) -> list[dict]:
        """AI (Dodil Models embeddings): cards closest in meaning to card N, across my boards, with a score."""
        who = actor(ctx)

        def go():
            pairs = svc.similar_cards(who, card, limit)
            return [{**v, "similarity": round(score, 3)}
                    for v, (_, score) in zip(cards_with_context(svc, [c for c, _ in pairs]), pairs)]
        return run(go)

    @server.tool(annotations=READ)
    def board_summary(board: str, ctx: Context) -> dict:
        """AI (Dodil Models): a short daily summary of a board: what moved, what is late, what is blocked."""
        who = actor(ctx)
        return run(lambda: {"board": board, "summary": svc.board_summary(who, board_id(who, board))})

    @server.tool(annotations=WRITE)
    def create_board(name: str, ctx: Context, workspace: str | None = None) -> dict:
        """Create a board with the lists To do, Doing and Done. `workspace` only if I am in several."""
        who = actor(ctx)
        return run(lambda: board_view(svc.create_board(who, workspace_id(who, workspace), name)))

    @server.tool(annotations=READ)
    def list_members(ctx: Context, workspace: str | None = None) -> list[dict]:
        """People in the workspace and their roles (owner, member, viewer)."""
        who = actor(ctx)
        return run(lambda: [member_view(m) for m in svc.list_members(who, workspace_id(who, workspace))])

    @server.tool(annotations=WRITE)
    def invite_member(email: str, ctx: Context, role: str = "member", workspace: str | None = None) -> dict:
        """Owner only: add a person by email as `member` (can edit) or `viewer` (read only)."""
        who = actor(ctx)
        return run(lambda: member_view(svc.invite_member(who, workspace_id(who, workspace), email, role)))

    return server