eduard/cobra-boards

Cobra Boards, team task boards you can run from Claude or ChatGPT. FastAPI + DataK3 + appid + Dodil Models + MCP.

A remix template · 1 remixes · Remix eduard/cobra-boards

app/boards/doors/mcp.py

"""The MCP door: the same use cases as the web routes, as tools Claude or ChatGPT can call.

Each tool acts as the person who connected (their gateway identity), so an agent sees exactly
the boards its person sees, and a viewer's agent can't change anything. Tools are actions,
never raw table access. They answer in short text with card numbers, so the agent can talk
about "#12 on Ops" the way the team does.
"""
from typing import Optional

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

from boards.application.use_cases import Boards, NotFound
from boards.domain.model import Conflict, Forbidden, Invalid
from boards.doors import text
from boards.doors.identity import person_from

INSTRUCTIONS = """Cobra Boards: the team's task boards (boards → lists → cards).
- Name things the way people do: a board by its title ("Ops"), a card by its number ("12" or
  "#12") plus the board when the person has several, a list by its name ("Doing").
- Start with my_boards or my_week. Use show_board to see a board's lists and cards.
- Writes act as the signed-in person: a viewer can read but every change is refused.
- If two boards or cards match, the tool says so: ask the person which one, never guess.
- Pass expected_version from the card you read when editing, so you never overwrite a teammate.
- AI suggestions (suggest_checklist) change nothing; add what the person chose.
- Comments can @mention a teammate (by email or the start of it); they get a notification.
- There is no delete: archive_card takes a card off the board and restore_card brings it back.
- Prompts: weekly_review, standup and plan_goal walk the person through common routines."""

READ = ToolAnnotations(readOnlyHint=True, openWorldHint=False)
WRITE = ToolAnnotations(readOnlyHint=False, destructiveHint=False, openWorldHint=False)
CHANGE = ToolAnnotations(readOnlyHint=False, destructiveHint=True, idempotentHint=False,
                         openWorldHint=False)


def build_mcp(boards: Boards) -> MCPServer:
    server = MCPServer(name="cobra-boards", 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 as e:
            raise ToolError(f"Not found: {e}") from None
        except Forbidden as e:
            raise ToolError(f"Not allowed: {e}") from None
        except Conflict as e:
            raise ToolError(f"Conflict: {e}") from None
        except Invalid as e:
            raise ToolError(str(e)) from None

    # ------------------------------------------------------------------ reading
    @server.tool(annotations=READ)
    def my_boards(ctx: Context) -> str:
        """The boards you are on, your role on each, and how many cards are open or overdue."""
        return text.boards(call(ctx, boards.my_boards))

    @server.tool(annotations=READ)
    def show_board(board: str, ctx: Context) -> str:
        """A board's lists and cards. `board` is its title (e.g. "Ops") or id."""
        return text.board(call(ctx, boards.show_board, board))

    @server.tool(annotations=READ)
    def my_week(ctx: Context, days: int = 7) -> str:
        """What's on my plate: my open cards across all my boards, grouped into overdue,
        due in the next `days` days, no due date, and later. Answers "what's on my plate
        this week?"."""
        return text.my_week(call(ctx, boards.my_cards, days))

    @server.tool(annotations=READ)
    def overdue(ctx: Context, board: Optional[str] = None) -> str:
        """Open cards past their due date, on one board or on all your boards."""
        items = call(ctx, boards.overdue, board)
        return text.card_list(items, "Nothing is overdue. 🎉", "Overdue")

    @server.tool(annotations=READ)
    def find_cards(query: str, ctx: Context, board: Optional[str] = None) -> str:
        """Find cards whose title, description or labels contain `query`."""
        items = call(ctx, boards.search_cards, query, board)
        return text.card_list(items, f"No card mentions “{query}”.", f"Cards matching “{query}”")

    @server.tool(annotations=READ)
    def show_card(card: str, ctx: Context, board: Optional[str] = None) -> str:
        """One card in full: description, people, due date, labels, what it waits on and what
        waits on it, its checklist and recent history. `card` is its number ("12") or id."""
        return text.card(call(ctx, boards.card_detail, card, board))

    @server.tool(annotations=READ)
    def card_history(card: str, ctx: Context, board: Optional[str] = None) -> str:
        """Every change to a card: who, when, and what changed (before → after)."""
        return text.history(call(ctx, boards.card_history, card, board))

    @server.tool(annotations=READ)
    def what_is_blocking(card: str, ctx: Context, board: Optional[str] = None) -> str:
        """Everything standing in the way of a card, followed down the chain of "blocked by"
        links, and which card to start with. Answers "what is blocking the launch?"."""
        return text.blocking(call(ctx, boards.what_is_blocking, card, board))

    # ------------------------------------------------------------------ boards & lists
    @server.tool(annotations=WRITE)
    def create_board(title: str, ctx: Context, template: str = "basic") -> str:
        """Create a board; you become its owner. template picks the starting lists:
        basic (To do / Doing / Done), sprint, launch, onboarding or support."""
        b = call(ctx, boards.create_board, title, None, template)
        return f"Created **{b['title']}** with lists {', '.join(b['lists'])}. You're the owner."

    @server.tool(annotations=WRITE)
    def add_list(board: str, name: str, ctx: Context, is_done: bool = False) -> str:
        """Add a list to a board. is_done=True means cards there count as finished."""
        lst = call(ctx, boards.add_list, board, name, is_done)
        return f"Added the list **{lst['name']}**" + (" (a done list)" if lst["is_done"] else "")

    @server.tool(annotations=CHANGE)
    def invite(board: str, email: str, ctx: Context, role: str = "member") -> str:
        """Owner only: give someone access to a board as 'member' (can edit) or 'viewer'
        (read only). Calling it again changes their role. They sign in with that email."""
        r = call(ctx, boards.invite, board, email, role)
        return f"{r['email']} is now a {r['role']} on {r['board']}."

    @server.tool(annotations=CHANGE)
    def remove_member(board: str, email: str, ctx: Context) -> str:
        """Owner only: take someone off a board. Their cards there become unassigned."""
        r = call(ctx, boards.remove_member, board, email)
        return (f"Removed {r['removed']} from {r['board']}; "
                f"{r['cards_unassigned']} card(s) unassigned.")

    # ------------------------------------------------------------------ cards
    @server.tool(annotations=WRITE)
    def add_card(board: str, title: str, ctx: Context, list: str = "",
                 description: str = "", assignee: str = "", due: str = "",
                 labels: Optional[list[str]] = None) -> str:
        """Add a card. Example: board="Ops", title="Send invoice", list="Doing".
        assignee is a teammate's email or "me". due takes YYYY-MM-DD, "today", "tomorrow",
        "next week" or "+3d"."""
        c = call(ctx, boards.add_card, board, title, list, description, assignee or None,
                 due or None, labels)
        return text.card_changed(c, "Added")

    @server.tool(annotations=CHANGE)
    def move_card(card: str, list: str, ctx: Context, board: Optional[str] = None,
                  expected_version: Optional[int] = None) -> str:
        """Move a card to another list, e.g. card="12", list="Done"."""
        c = call(ctx, boards.move_card, card, list, board, None, expected_version)
        return text.card_changed(c, "Moved")

    @server.tool(annotations=CHANGE)
    def assign_card(card: str, ctx: Context, assignee: str = "",
                    board: Optional[str] = None, expected_version: Optional[int] = None) -> str:
        """Assign a card to someone on its board (their email, or "me"), or "" to unassign."""
        c = call(ctx, boards.update_card, card, board, expected_version,
                 assignee=assignee or None)
        return text.card_changed(c, "Assigned" if assignee else "Unassigned")

    @server.tool(annotations=CHANGE)
    def edit_card(card: str, ctx: Context, board: Optional[str] = None,
                  expected_version: Optional[int] = None, title: Optional[str] = None,
                  description: Optional[str] = None, due: Optional[str] = None,
                  labels: Optional[list[str]] = None) -> str:
        """Change a card's title, description, due date (or "none" to clear) or labels.
        Only what you pass changes."""
        kwargs = {"title": title, "description": description, "labels": labels}
        if due is not None:
            kwargs["due_date"] = due
        c = call(ctx, boards.update_card, card, board, expected_version, **kwargs)
        return text.card_changed(c, "Updated")

    @server.tool(annotations=WRITE)
    def add_blocker(card: str, blocked_by: str, ctx: Context,
                    board: Optional[str] = None) -> str:
        """Record that a card can't finish before another one: card="3", blocked_by="2"."""
        r = call(ctx, boards.add_blocker, card, blocked_by, board)
        return (f"#{r['card']} already waits on #{r['blocked_by']}." if r.get("already")
                else f"#{r['card']} is now blocked by #{r['blocked_by']}.")

    @server.tool(annotations=CHANGE)
    def remove_blocker(card: str, blocked_by: str, ctx: Context,
                       board: Optional[str] = None) -> str:
        """Remove a "blocked by" link between two cards."""
        r = call(ctx, boards.remove_blocker, card, blocked_by, board)
        return f"#{r['card']} no longer waits on #{r['unblocked_from']}."

    # ------------------------------------------------------------------ checklists & AI
    @server.tool(annotations=READ)
    def suggest_checklist(card: str, ctx: Context, board: Optional[str] = None) -> str:
        """AI (Dodil Models) suggests a checklist for a card. Saves NOTHING: show the steps to
        the person and add the ones they want with add_checklist_items."""
        r = call(ctx, boards.suggest_checklist, card, board)
        return (f"Suggested steps for #{r['card']} {r['title']} (not saved):\n"
                + "\n".join(f"- {s}" for s in r["suggested_steps"]))

    @server.tool(annotations=WRITE)
    def add_checklist_items(card: str, items: list[str], ctx: Context,
                            board: Optional[str] = None) -> str:
        """Add steps to a card's checklist."""
        return text.checklist(call(ctx, boards.add_checklist_items, card, items, board))

    @server.tool(annotations=CHANGE)
    def check_item(card: str, item_id: str, ctx: Context, done: bool = True,
                   board: Optional[str] = None) -> str:
        """Tick (or untick with done=False) a checklist step; ids are shown by show_card."""
        return text.checklist(call(ctx, boards.set_checklist_item, card, item_id, done,
                                   None, False, board))

    @server.tool(annotations=READ)
    def similar_cards(ctx: Context, card: Optional[str] = None, text_query: Optional[str] = None,
                      board: Optional[str] = None) -> str:
        """Cards that mean the same thing even in different words (meaning search over your
        boards only). Give a card number, or describe it with text_query."""
        return text.similar(call(ctx, boards.similar_cards, card, text_query, board))

    # ------------------------------------------------------------------ comments & notifications
    @server.tool(annotations=WRITE)
    def add_comment(card: str, text: str, ctx: Context, board: Optional[str] = None) -> str:
        """Comment on a card. Write @name (a teammate's email, or its start) to mention them:
        they get a notification. The card's assignee is told too."""
        r = call(ctx, boards.add_comment, card, text, board)
        who = (" Mentioned " + ", ".join(r["mentioned"]) + ".") if r["mentioned"] else ""
        return f"Commented on #{r['card']}.{who} {r['notified']} person(s) notified."

    @server.tool(annotations=READ)
    def my_notifications(ctx: Context, unread_only: bool = True) -> str:
        """What changed for me: @mentions, cards assigned to me, comments on my cards."""
        return text.notifications(call(ctx, boards.my_notifications, unread_only))

    @server.tool(annotations=WRITE)
    def mark_notifications_read(ctx: Context, ids: Optional[list[str]] = None) -> str:
        """Mark notifications as read: the given ids, or all of them when ids is empty."""
        r = call(ctx, boards.mark_notifications_read, ids)
        return f"Marked {r['marked']} notification(s) as read."

    # ------------------------------------------------------------------ archive
    @server.tool(annotations=CHANGE)
    def archive_card(card: str, ctx: Context, board: Optional[str] = None) -> str:
        """Take a card off the board (there is no delete). It keeps its number, history and
        comments, and restore_card brings it back."""
        r = call(ctx, boards.archive_card, card, board)
        return (f"#{r['card']} {r['title']} was already archived." if r.get("already")
                else f"Archived #{r['card']} {r['title']}. restore_card brings it back.")

    @server.tool(annotations=WRITE)
    def restore_card(card: str, ctx: Context, board: Optional[str] = None) -> str:
        """Bring an archived card back to its list."""
        r = call(ctx, boards.restore_card, card, board)
        return f"Restored #{r['card']} {r['title']} to {r['list']}."

    @server.tool(annotations=READ)
    def archived_cards(board: str, ctx: Context) -> str:
        """The cards archived on a board (off the board, not deleted); restore_card brings one back."""
        items = call(ctx, boards.archived_cards, board)
        if not items:
            return "Nothing is archived on this board."
        return "Archived:\n" + "\n".join(f"- #{c['number']} {c['title']}" for c in items)

    # ------------------------------------------------------------------ prompts
    @server.prompt(title="Weekly review")
    def weekly_review(board: str) -> str:
        """Review a board's week: what finished, what's late, what's stuck, what to do next."""
        return (f"Run a weekly review of the board \"{board}\" with me. Use the Cobra tools: "
                f"daily_summary and show_board for \"{board}\", overdue, and what_is_blocking for "
                "every blocked card. Then tell me, in this order: what finished, what is late and "
                "who has it, what is stuck and the card to start with, and 3 concrete next steps. "
                "Don't change anything; ask me before any move or edit.")

    @server.prompt(title="Stand-up")
    def standup(board: str) -> str:
        """A 2-minute stand-up for me on one board."""
        return (f"Give me my stand-up for \"{board}\". Use my_week, my_notifications and "
                f"show_board for \"{board}\". Three short parts: what I moved since yesterday "
                "(from card_history), what I'm on today (due or overdue first), and what is "
                "blocking me. Then ask if I want to move or comment on anything.")

    @server.prompt(title="Plan a goal")
    def plan_goal(board: str, goal: str) -> str:
        """Turn a goal into cards on a board, with my approval first."""
        return (f"Help me plan this goal on the board \"{board}\": {goal}\n"
                "First show me a short plan: 4 to 8 cards, each with a title, a list and, if it "
                "waits on another, its blocker. Ask me to confirm or change it. Only after I "
                "say yes, create the cards with add_card and link them with add_blocker. "
                "Never guess an assignee; ask.")

    @server.tool(annotations=READ)
    def daily_summary(board: str, ctx: Context) -> str:
        """A short morning briefing of a board: what's at risk, due today, and progress.
        The counts come from the board itself; the AI only words them."""
        r = call(ctx, boards.daily_summary, board)
        return f"**{r['board']}** · {r['date']}\n{r['summary']}"

    return server