mobiledev/taskflow
TaskFlow, a team task board you run from the web or from Claude and ChatGPT. FastAPI + DataK3 + appid + Models + MCP.
A remix template · 0 remixes · Remix mobiledev/taskflow
app/taskflow/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 assistant sees exactly
the boards its person sees. Tools are actions, never raw table access. Boards can be named by
title ("Ops") and cards by number ("12" or "#12"), the way people talk.
"""
from typing import Optional
from mcp.server.mcpserver import Context, MCPServer
from mcp.server.mcpserver.exceptions import ToolError
from taskflow.application.use_cases import NotFound, TaskFlow
from taskflow.doors.identity import person_from
from taskflow.domain.model import Forbidden, Invalid
INSTRUCTIONS = (
"TaskFlow, a team task board. A board has lists (usually 'To do', 'Doing', 'Done'; the list "
"called Done means finished) holding numbered cards. Name a board by its title or id and a card by "
"its number like '12' (add board when two boards share a number). For 'what's on my plate "
"this week' use my_cards with due_within_days=7; for 'what's overdue' use overdue. Dates "
"are YYYY-MM-DD. Viewers can read but not change anything.")
def _cards(cards: list) -> dict:
"""One object per answer, so "nothing is overdue" reads as {"count": 0}, not silence."""
return {"count": len(cards), "cards": cards}
def build_mcp(app: TaskFlow) -> MCPServer:
server = MCPServer(name="taskflow", 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 e:
raise ToolError(str(e))
# --- reading -------------------------------------------------------------------------
@server.tool()
def my_boards(ctx: Context) -> dict:
"""The boards you are on, with your role on each."""
return {"boards": call(ctx, app.my_boards)}
@server.tool()
def show_board(board: str, ctx: Context) -> dict:
"""A board with its lists, cards (in order) and members."""
return call(ctx, app.show_board, board)
@server.tool()
def show_card(card: str, ctx: Context, board: Optional[str] = None) -> dict:
"""One card with all its fields and its recent history."""
return call(ctx, app.show_card, card, board)
@server.tool()
def my_cards(ctx: Context, due_within_days: Optional[int] = None) -> dict:
"""Open cards assigned to you on all your boards, soonest due first. Use
due_within_days=7 for "what's on my plate this week?" (overdue cards included)."""
return _cards(call(ctx, app.my_cards, due_within_days))
@server.tool()
def overdue(ctx: Context, board: Optional[str] = None) -> dict:
"""Cards past their due date and not finished, on one board or all of yours."""
return _cards(call(ctx, app.overdue, board))
@server.tool()
def card_history(card: str, ctx: Context, board: Optional[str] = None) -> dict:
"""Every change to a card, newest first: who, when, what changed."""
return {"history": call(ctx, app.card_history, card, board)}
@server.tool()
def board_activity(board: str, ctx: Context, limit: int = 30) -> dict:
"""Recent changes across a whole board, newest first."""
return {"activity": call(ctx, app.board_activity, board, limit)}
@server.tool()
def what_blocks(card: str, ctx: Context, board: Optional[str] = None) -> dict:
"""Everything still open that a card waits on, directly or down the chain
("what is blocking the launch?")."""
return call(ctx, app.what_blocks, card, board)
# --- changing ------------------------------------------------------------------------
@server.tool()
def create_board(title: str, ctx: Context, lists: Optional[list[str]] = None) -> dict:
"""Create a board; you become its owner. Lists default to To do, Doing, Done."""
return call(ctx, app.create_board, title, lists)
@server.tool()
def add_card(board: str, title: str, ctx: Context, list_name: Optional[str] = None,
description: str = "", assignee: Optional[str] = None,
due_date: Optional[str] = None, labels: Optional[list[str]] = None) -> dict:
"""Add a card, e.g. 'add a card Send invoice to Doing on the Ops board' is
add_card(board="Ops", title="Send invoice", list_name="Doing").
board: board title or id. list_name: the list to put it in (default: the first list).
assignee: a member's email. due_date: YYYY-MM-DD. labels: e.g. ["billing"]."""
return call(ctx, app.add_card, board, title, list_name, description, assignee,
due_date, labels)
@server.tool()
def move_card(card: str, list_name: str, ctx: Context, board: Optional[str] = None) -> dict:
"""Move a card to another list, e.g. 'move card 12 to Done' is
move_card(card="12", list_name="Done"). Add board when two boards have a #12."""
return call(ctx, app.move_card, card, list_name, board)
@server.tool()
def assign_card(card: str, ctx: Context, assignee: Optional[str] = None,
board: Optional[str] = None) -> dict:
"""Give a card to a member by email, or leave assignee empty to unassign."""
return call(ctx, app.assign_card, card, assignee, board)
@server.tool()
def update_card(card: str, ctx: Context, board: Optional[str] = None,
title: Optional[str] = None, description: Optional[str] = None,
due_date: Optional[str] = None, labels: Optional[list[str]] = None,
clear_due_date: bool = False) -> dict:
"""Change a card's title, description, due date (YYYY-MM-DD) or labels. Only the
fields you pass change."""
changes = {k: v for k, v in (("title", title), ("description", description),
("due_date", due_date), ("labels", labels)) if v is not None}
if clear_due_date:
changes["due_date"] = None
return call(ctx, app.update_card, card, board, **changes)
@server.tool()
def block_card(card: str, blocked_by: str, ctx: Context, board: Optional[str] = None
) -> dict:
"""Record that `card` can't finish before `blocked_by` (both on the same board)."""
return call(ctx, app.block_card, card, blocked_by, board)
@server.tool()
def unblock_card(card: str, blocked_by: str, ctx: Context, board: Optional[str] = None
) -> dict:
"""Remove a 'blocked by' link."""
return call(ctx, app.unblock_card, card, blocked_by, board)
@server.tool()
def add_list(board: str, name: str, ctx: Context, before: Optional[str] = None) -> dict:
"""Owner only: add a list to a board. It goes before Done unless `before` names
another list."""
return call(ctx, app.add_list, board, name, before)
@server.tool()
def rename_list(board: str, name: str, new_name: str, ctx: Context) -> dict:
"""Owner only: rename a list; its cards move with it."""
return call(ctx, app.rename_list, board, name, new_name)
@server.tool()
def delete_list(board: str, name: str, ctx: Context) -> dict:
"""Owner only: delete an EMPTY list (move its cards out first). Deleting a whole board
is only possible from the web page, on purpose."""
return call(ctx, app.delete_list, board, name)
@server.tool()
def delete_card(card: str, ctx: Context, board: Optional[str] = None) -> dict:
"""Delete a card for good. Confirm with the person first. It stays in the history."""
return call(ctx, app.delete_card, card, board)
@server.tool()
def add_member(board: str, email: str, ctx: Context, role: str = "member") -> dict:
"""Owner only: invite someone as 'member' or 'viewer', or change their role."""
return call(ctx, app.add_member, board, email, role)
@server.tool()
def remove_member(board: str, email: str, ctx: Context) -> dict:
"""Owner only: take someone off a board; their cards become unassigned."""
return call(ctx, app.remove_member, board, email)
@server.tool()
def add_checklist(card: str, items: list[str], ctx: Context, board: Optional[str] = None
) -> dict:
"""Add checklist items to a card."""
return call(ctx, app.add_checklist, card, items, board)
@server.tool()
def tick(card: str, item: int, ctx: Context, done: bool = True,
board: Optional[str] = None) -> dict:
"""Tick (or untick with done=false) checklist item number `item` (1 = first)."""
return call(ctx, app.tick, card, item, done, board)
# --- AI (Dodil Models) ---------------------------------------------------------------
@server.tool()
def checklist_from_ai(card: str, ctx: Context, board: Optional[str] = None) -> dict:
""""Turn this card into a checklist": AI writes the steps and adds them to the card."""
return call(ctx, app.checklist_from_ai, card, board)
@server.tool()
def similar_cards(card: str, ctx: Context, board: Optional[str] = None, limit: int = 5
) -> dict:
"""Cards on the same board that mean something close (search by meaning)."""
return _cards(call(ctx, app.similar_cards, card, board, limit))
@server.tool()
def board_summary(board: str, ctx: Context) -> dict:
"""A short daily summary of a board: counts, overdue work, what changed."""
return call(ctx, app.board_summary, board)
return server