538 lines
17 KiB
Python
538 lines
17 KiB
Python
"""Todo MCP server — lightweight JSON-backed task management.
|
|
|
|
Exposes ``todo_*`` tools for listing, creating, updating, completing and
|
|
deleting tasks. Follows the same style as the Harrier / OSINT-MCP servers:
|
|
flattened, action-oriented tool names and individual parameters.
|
|
|
|
Persistence lives in ``store.py`` (a single JSON file; default
|
|
``~/.todo-mcp/todos.json``, override with ``TODO_MCP_STORE``).
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import uuid
|
|
from datetime import date as date_cls
|
|
from typing import List, Optional, Union
|
|
|
|
from mcp.server.fastmcp import FastMCP
|
|
from pydantic import BaseModel, ConfigDict, Field, field_validator
|
|
|
|
from .store import Store, now_iso
|
|
|
|
mcp = FastMCP("todo-mcp")
|
|
|
|
PRIORITIES = ("urgent", "high", "medium", "low")
|
|
PRIOR_ORDER = {p: i for i, p in enumerate(PRIORITIES)}
|
|
STATUSES = ("active", "done")
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# Helpers
|
|
# --------------------------------------------------------------------------- #
|
|
def _store() -> Store:
|
|
return Store()
|
|
|
|
|
|
def _new_id() -> str:
|
|
return "t_" + uuid.uuid4().hex[:8]
|
|
|
|
|
|
def _find(tasks: List[dict], tid: str) -> Optional[dict]:
|
|
for t in tasks:
|
|
if t.get("id") == tid:
|
|
return t
|
|
return None
|
|
|
|
|
|
def _sort(tasks: List[dict]) -> List[dict]:
|
|
return sorted(
|
|
tasks,
|
|
key=lambda t: (
|
|
PRIOR_ORDER.get(t.get("priority"), len(PRIOR_ORDER)),
|
|
t.get("due") or "z",
|
|
t.get("created_at") or "",
|
|
),
|
|
)
|
|
|
|
|
|
def _fmt(item: dict, fmt: str) -> str:
|
|
if fmt == "markdown":
|
|
return _task_markdown(item)
|
|
return json.dumps(item, indent=2)
|
|
|
|
|
|
def _task_markdown(t: dict) -> str:
|
|
box = "[x]" if t.get("status") == "done" else "[ ]"
|
|
line = f"- {box} `{t['id']}`"
|
|
if t.get("priority"):
|
|
line += f" 🔴 {t['priority']}"
|
|
line += f" **{t['title']}**"
|
|
if t.get("due"):
|
|
line += f" 📅 {t['due']}"
|
|
meta = []
|
|
if t.get("project"):
|
|
meta.append(f"📁 {t['project']}")
|
|
if t.get("tags"):
|
|
meta.append("🏷️ " + ", ".join(t["tags"]))
|
|
if meta:
|
|
line += " · " + " · ".join(meta)
|
|
out = [line]
|
|
if t.get("description"):
|
|
out.extend((" " + ln).rstrip() for ln in t["description"].splitlines())
|
|
return "\n".join(out)
|
|
|
|
|
|
def _list_markdown(tasks: List[dict]) -> str:
|
|
if not tasks:
|
|
return "_No tasks._"
|
|
active = _sort([t for t in tasks if t.get("status") == "active"])
|
|
done = _sort([t for t in tasks if t.get("status") == "done"])
|
|
lines = ["## Active", ""]
|
|
lines.extend(_task_markdown(t) for t in active) or ["_None._"]
|
|
lines.append("")
|
|
lines.append("## Done")
|
|
lines.append("")
|
|
lines.extend(_task_markdown(t) for t in done) or ["_None._"]
|
|
return "\n".join(lines)
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# Validation models (single source of truth for coercion / constraints)
|
|
# --------------------------------------------------------------------------- #
|
|
class _Base(BaseModel):
|
|
model_config = ConfigDict(str_strip_whitespace=True, extra="forbid")
|
|
|
|
|
|
class AddInput(_Base):
|
|
title: str = Field(..., min_length=1, max_length=500, description="Task title.")
|
|
description: Optional[str] = Field(None, max_length=5000, description="Optional description.")
|
|
priority: Optional[str] = Field(None, description="urgent | high | medium | low (default medium).")
|
|
due: Optional[str] = Field(None, description="Optional due date, YYYY-MM-DD.")
|
|
tags: Optional[List[str]] = Field(None, description="Tags (list or comma-separated string).")
|
|
project: Optional[str] = Field(None, max_length=100, description="Project/group name.")
|
|
format: str = Field("json", description="Output format: json (default) or markdown.")
|
|
|
|
@field_validator("priority", mode="before")
|
|
@classmethod
|
|
def _norm_priority(cls, v: Optional[str]) -> Optional[str]:
|
|
if v is None:
|
|
return None
|
|
v = str(v).strip().lower()
|
|
if v not in PRIORITIES:
|
|
raise ValueError(f"priority must be one of {PRIORITIES}")
|
|
return v
|
|
|
|
@field_validator("due", mode="before")
|
|
@classmethod
|
|
def _norm_due(cls, v: Optional[str]) -> Optional[str]:
|
|
if v is None or str(v).strip() == "":
|
|
return None
|
|
try:
|
|
return date_cls.fromisoformat(str(v)).isoformat()
|
|
except ValueError:
|
|
raise ValueError("due must be an ISO date (YYYY-MM-DD)")
|
|
|
|
@field_validator("tags", mode="before")
|
|
@classmethod
|
|
def _norm_tags(cls, v: Optional[object]) -> Optional[List[str]]:
|
|
if v is None:
|
|
return None
|
|
if isinstance(v, str):
|
|
v = v.split(",")
|
|
return [str(x).strip() for x in v if str(x).strip()]
|
|
|
|
|
|
class ListInput(_Base):
|
|
status: Optional[str] = Field(None, description="Filter: active | done.")
|
|
priority: Optional[str] = Field(None, description="Filter by priority.")
|
|
project: Optional[str] = Field(None, description="Filter by project name.")
|
|
tag: Optional[str] = Field(None, description="Filter by a single tag.")
|
|
limit: int = Field(50, ge=1, le=500, description="Max results to return.")
|
|
offset: int = Field(0, ge=0, description="Results to skip (pagination).")
|
|
format: str = Field("json", description="Output format: json (default) or markdown.")
|
|
|
|
@field_validator("status", "priority", "tag", mode="before")
|
|
@classmethod
|
|
def _norm_filter(cls, v: Optional[str]) -> Optional[str]:
|
|
if v is None:
|
|
return None
|
|
v = str(v).strip().lower()
|
|
return v or None
|
|
|
|
@field_validator("priority", mode="before")
|
|
@classmethod
|
|
def _norm_pri(cls, v: Optional[str]) -> Optional[str]:
|
|
if v is None:
|
|
return None
|
|
if v in PRIORITIES:
|
|
return v
|
|
raise ValueError(f"priority must be one of {PRIORITIES}")
|
|
|
|
|
|
class GetInput(_Base):
|
|
id: str = Field(..., min_length=1, description="Task id (e.g. t_1a2b3c4d).")
|
|
format: str = Field("json", description="Output format: json (default) or markdown.")
|
|
|
|
|
|
class UpdateInput(_Base):
|
|
id: str = Field(..., min_length=1, description="Task id to update.")
|
|
title: Optional[str] = Field(None, min_length=1, max_length=500, description="New title.")
|
|
description: Optional[str] = Field(None, max_length=5000, description="New description.")
|
|
priority: Optional[str] = Field(None, description="urgent | high | medium | low.")
|
|
due: Optional[str] = Field(None, description="New due date, YYYY-MM-DD.")
|
|
tags: Optional[List[str]] = Field(None, description="Replace tags (list or comma string).")
|
|
project: Optional[str] = Field(None, max_length=100, description="New project name.")
|
|
status: Optional[str] = Field(None, description="New status: active | done.")
|
|
format: str = Field("json", description="Output format: json (default) or markdown.")
|
|
|
|
@field_validator("priority", "status", mode="before")
|
|
@classmethod
|
|
def _norm_update(cls, v: Optional[str]) -> Optional[str]:
|
|
if v is None:
|
|
return None
|
|
v = str(v).strip().lower()
|
|
if v in PRIORITIES or v in STATUSES:
|
|
return v
|
|
raise ValueError(f"value must be one of {PRIORITIES + STATUSES}")
|
|
|
|
@field_validator("due", mode="before")
|
|
@classmethod
|
|
def _norm_due(cls, v: Optional[str]) -> Optional[str]:
|
|
if v is None or str(v).strip() == "":
|
|
return None
|
|
try:
|
|
return date_cls.fromisoformat(str(v)).isoformat()
|
|
except ValueError:
|
|
raise ValueError("due must be an ISO date (YYYY-MM-DD)")
|
|
|
|
@field_validator("tags", mode="before")
|
|
@classmethod
|
|
def _norm_tags(cls, v: Optional[object]) -> Optional[List[str]]:
|
|
if v is None:
|
|
return None
|
|
if isinstance(v, str):
|
|
v = v.split(",")
|
|
return [str(x).strip() for x in v if str(x).strip()]
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# Tools (flattened parameters, matching the Harrier convention)
|
|
#
|
|
# Plain Python defaults are used on the signatures so the tools work whether
|
|
# called directly or through FastMCP; the Pydantic models above do all the
|
|
# real validation and coercion.
|
|
# --------------------------------------------------------------------------- #
|
|
@mcp.tool(
|
|
name="todo_add",
|
|
annotations={
|
|
"title": "Add a task",
|
|
"readOnlyHint": False,
|
|
"destructiveHint": False,
|
|
"idempotentHint": False,
|
|
"openWorldHint": False,
|
|
},
|
|
)
|
|
async def todo_add(
|
|
title: str,
|
|
description: Optional[str] = None,
|
|
priority: Optional[str] = None,
|
|
due: Optional[str] = None,
|
|
tags: Optional[Union[str, List[str]]] = None,
|
|
project: Optional[str] = None,
|
|
format: str = "json",
|
|
) -> str:
|
|
'''Create a new task and persist it.
|
|
|
|
Args:
|
|
title (str): Task title (required).
|
|
description (Optional[str]): Optional longer description.
|
|
priority (Optional[str]): urgent | high | medium | low (default medium).
|
|
due (Optional[str]): Optional due date, YYYY-MM-DD.
|
|
tags (Optional[Union[str, list[str]]]): Tags; list or comma-separated string.
|
|
project (Optional[str]): Optional project/group name.
|
|
format (str): json (default) or markdown.
|
|
|
|
Returns:
|
|
str: The created task as JSON, or markdown if format="markdown".
|
|
'''
|
|
params = AddInput(title=title, description=description, priority=priority, due=due,
|
|
tags=tags, project=project, format=format)
|
|
tasks = _store().load()
|
|
task = {
|
|
"id": _new_id(),
|
|
"title": params.title,
|
|
"description": params.description,
|
|
"priority": params.priority or "medium",
|
|
"status": "active",
|
|
"due": params.due,
|
|
"tags": params.tags or [],
|
|
"project": params.project,
|
|
"created_at": now_iso(),
|
|
"updated_at": now_iso(),
|
|
}
|
|
tasks.append(task)
|
|
_store().save(tasks)
|
|
return _fmt(task, params.format)
|
|
|
|
|
|
@mcp.tool(
|
|
name="todo_list",
|
|
annotations={
|
|
"title": "List tasks",
|
|
"readOnlyHint": True,
|
|
"destructiveHint": False,
|
|
"idempotentHint": True,
|
|
"openWorldHint": False,
|
|
},
|
|
)
|
|
async def todo_list(
|
|
status: Optional[str] = None,
|
|
priority: Optional[str] = None,
|
|
project: Optional[str] = None,
|
|
tag: Optional[str] = None,
|
|
limit: int = 50,
|
|
offset: int = 0,
|
|
format: str = "json",
|
|
) -> str:
|
|
'''List tasks with optional filters and pagination.
|
|
|
|
Args:
|
|
status (Optional[str]): active | done.
|
|
priority (Optional[str]): urgent | high | medium | low.
|
|
project (Optional[str]): project/group name.
|
|
tag (Optional[str]): single tag.
|
|
limit (int): 1-500 (default 50).
|
|
offset (int): results to skip (default 0).
|
|
format (str): json (default) or markdown.
|
|
|
|
Returns:
|
|
str: JSON {total, count, offset, tasks} or a markdown grouping.
|
|
'''
|
|
params = ListInput(status=status, priority=priority, project=project, tag=tag,
|
|
limit=limit, offset=offset, format=format)
|
|
tasks = _store().load()
|
|
if params.status:
|
|
tasks = [t for t in tasks if t.get("status") == params.status]
|
|
if params.priority:
|
|
tasks = [t for t in tasks if t.get("priority") == params.priority]
|
|
if params.project:
|
|
tasks = [t for t in tasks if t.get("project") == params.project]
|
|
if params.tag:
|
|
tasks = [t for t in tasks if params.tag in (t.get("tags") or [])]
|
|
tasks = _sort(tasks)
|
|
page = tasks[params.offset:params.offset + params.limit]
|
|
if params.format == "markdown":
|
|
return _list_markdown(page)
|
|
return json.dumps(
|
|
{"total": len(tasks), "count": len(page), "offset": params.offset, "tasks": page},
|
|
indent=2,
|
|
)
|
|
|
|
|
|
@mcp.tool(
|
|
name="todo_get",
|
|
annotations={
|
|
"title": "Get a task",
|
|
"readOnlyHint": True,
|
|
"destructiveHint": False,
|
|
"idempotentHint": True,
|
|
"openWorldHint": False,
|
|
},
|
|
)
|
|
async def todo_get(
|
|
id: str,
|
|
format: str = "json",
|
|
) -> str:
|
|
'''Fetch a single task by id.
|
|
|
|
Args:
|
|
id (str): Task id.
|
|
format (str): json (default) or markdown.
|
|
|
|
Returns:
|
|
str: The task as JSON, or an error string if not found.
|
|
'''
|
|
params = GetInput(id=id, format=format)
|
|
task = _find(_store().load(), params.id)
|
|
if not task:
|
|
return f"Error: no todo with id '{params.id}'"
|
|
return _fmt(task, params.format)
|
|
|
|
|
|
@mcp.tool(
|
|
name="todo_update",
|
|
annotations={
|
|
"title": "Update a task",
|
|
"readOnlyHint": False,
|
|
"destructiveHint": False,
|
|
"idempotentHint": True,
|
|
"openWorldHint": False,
|
|
},
|
|
)
|
|
async def todo_update(
|
|
id: str,
|
|
title: Optional[str] = None,
|
|
description: Optional[str] = None,
|
|
priority: Optional[str] = None,
|
|
due: Optional[str] = None,
|
|
tags: Optional[Union[str, List[str]]] = None,
|
|
project: Optional[str] = None,
|
|
status: Optional[str] = None,
|
|
format: str = "json",
|
|
) -> str:
|
|
'''Update mutable fields on an existing task.
|
|
|
|
Only the fields you pass are changed; omitted fields are left as-is.
|
|
Pass tags to replace the whole tag set.
|
|
|
|
Args:
|
|
id (str): Task id.
|
|
title (Optional[str]): New title.
|
|
description (Optional[str]): New description.
|
|
priority (Optional[str]): urgent|high|medium|low.
|
|
due (Optional[str]): New due date, YYYY-MM-DD.
|
|
tags (Optional[Union[str, list[str]]]): Replace tags.
|
|
project (Optional[str]): New project name.
|
|
status (Optional[str]): active | done.
|
|
format (str): json (default) or markdown.
|
|
|
|
Returns:
|
|
str: The updated task as JSON, or an error string if not found.
|
|
'''
|
|
params = UpdateInput(id=id, title=title, description=description, priority=priority,
|
|
due=due, tags=tags, project=project, status=status, format=format)
|
|
store = _store()
|
|
tasks = store.load()
|
|
task = _find(tasks, params.id)
|
|
if not task:
|
|
return f"Error: no todo with id '{params.id}'"
|
|
changes = {
|
|
k: getattr(params, k)
|
|
for k in ("title", "description", "priority", "due", "tags", "project", "status")
|
|
if getattr(params, k) is not None
|
|
}
|
|
task.update(changes)
|
|
task["updated_at"] = now_iso()
|
|
store.save(tasks)
|
|
return _fmt(task, params.format)
|
|
|
|
|
|
@mcp.tool(
|
|
name="todo_toggle",
|
|
annotations={
|
|
"title": "Toggle task status",
|
|
"readOnlyHint": False,
|
|
"destructiveHint": False,
|
|
"idempotentHint": True,
|
|
"openWorldHint": False,
|
|
},
|
|
)
|
|
async def todo_toggle(id: str) -> str:
|
|
'''Flip a task between active and done (quick complete/uncomplete).
|
|
|
|
Args:
|
|
id (str): Task id to toggle.
|
|
|
|
Returns:
|
|
str: The updated task as JSON, or an error string if not found.
|
|
'''
|
|
store = _store()
|
|
tasks = store.load()
|
|
task = _find(tasks, id)
|
|
if not task:
|
|
return f"Error: no todo with id '{id}'"
|
|
task["status"] = "done" if task.get("status") == "active" else "active"
|
|
task["updated_at"] = now_iso()
|
|
store.save(tasks)
|
|
return _fmt(task, "json")
|
|
|
|
|
|
@mcp.tool(
|
|
name="todo_delete",
|
|
annotations={
|
|
"title": "Delete a task",
|
|
"readOnlyHint": False,
|
|
"destructiveHint": True,
|
|
"idempotentHint": True,
|
|
"openWorldHint": False,
|
|
},
|
|
)
|
|
async def todo_delete(id: str) -> str:
|
|
'''Permanently delete a task by id.
|
|
|
|
Args:
|
|
id (str): Task id to delete.
|
|
|
|
Returns:
|
|
str: JSON confirming deletion, or an error string if not found.
|
|
'''
|
|
store = _store()
|
|
tasks = store.load()
|
|
if _find(tasks, id) is None:
|
|
return f"Error: no todo with id '{id}'"
|
|
store.save([t for t in tasks if t.get("id") != id])
|
|
return json.dumps({"deleted": id})
|
|
|
|
|
|
@mcp.tool(
|
|
name="todo_clear",
|
|
annotations={
|
|
"title": "Clear completed tasks",
|
|
"readOnlyHint": False,
|
|
"destructiveHint": True,
|
|
"idempotentHint": True,
|
|
"openWorldHint": False,
|
|
},
|
|
)
|
|
async def todo_clear(format: str = "json") -> str:
|
|
'''Remove all completed (done) tasks.
|
|
|
|
Args:
|
|
format (str): json (default) or markdown.
|
|
|
|
Returns:
|
|
str: JSON {cleared, remaining} counts.
|
|
'''
|
|
store = _store()
|
|
tasks = store.load()
|
|
keep = [t for t in tasks if t.get("status") != "done"]
|
|
removed = len(tasks) - len(keep)
|
|
store.save(keep)
|
|
return json.dumps({"cleared": removed, "remaining": len(keep)}, indent=2)
|
|
|
|
|
|
@mcp.tool(
|
|
name="todo_stats",
|
|
annotations={
|
|
"title": "Task summary",
|
|
"readOnlyHint": True,
|
|
"destructiveHint": False,
|
|
"idempotentHint": True,
|
|
"openWorldHint": False,
|
|
},
|
|
)
|
|
async def todo_stats() -> str:
|
|
'''Summarize the todo list: totals, by status, by priority, by project.
|
|
|
|
Returns:
|
|
str: JSON summary object.
|
|
'''
|
|
tasks = _store().load()
|
|
summary = {
|
|
"total": len(tasks),
|
|
"active": sum(1 for t in tasks if t.get("status") == "active"),
|
|
"done": sum(1 for t in tasks if t.get("status") == "done"),
|
|
"by_priority": {p: sum(1 for t in tasks if t.get("priority") == p) for p in PRIORITIES},
|
|
"projects": sorted({t.get("project") for t in tasks if t.get("project")}),
|
|
}
|
|
return json.dumps(summary, indent=2)
|
|
|
|
|
|
def _cli() -> None:
|
|
"""Entrypoint: run the stdio MCP server."""
|
|
mcp.run()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
_cli()
|