mirror of
https://github.com/kuhyx/todo-app.git
synced 2026-11-11 13:47:45 +01:00
Add read-only MCP server for the backlog
New standalone Python package under mcp/ that exposes the todo app's exported BACKLOG.md (NotesMarkdown format) as read-only MCP tools, so Claude Code and its subagents can query the backlog. The app is Dart/Flutter with no Python and no CLI; this server reads BACKLOG.md and never touches the sqlite_crdt DB (CRDT writes would corrupt cross-device merge). - todo_mcp/_backlog.py: pure parse_backlog(text) -> list[Note]. - todo_mcp/_mcp.py: FastMCP stdio server. Read-only tools list_backlog, get_note, backlog_stats; every result carries the file mtime so a stale export is visible; missing file returns a graceful "export it" message. Logs to stderr; no writes; no secrets (backlog is note text). - .mcp.json registers it via a dedicated ~/.venvs/todo-mcp interpreter. - 100% coverage; verified end-to-end over stdio; parses the real BACKLOG.md. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01LQGQj8Ne6TriBExv91nG8v
This commit is contained in:
parent
166d9f00d0
commit
0823e33d72
10
.mcp.json
Normal file
10
.mcp.json
Normal file
@ -0,0 +1,10 @@
|
||||
{
|
||||
"mcpServers": {
|
||||
"todo": {
|
||||
"type": "stdio",
|
||||
"command": "${HOME}/.venvs/todo-mcp/bin/python",
|
||||
"args": ["-m", "todo_mcp._mcp"],
|
||||
"env": {}
|
||||
}
|
||||
}
|
||||
}
|
||||
6
mcp/.gitignore
vendored
Normal file
6
mcp/.gitignore
vendored
Normal file
@ -0,0 +1,6 @@
|
||||
__pycache__/
|
||||
*.egg-info/
|
||||
*.pyc
|
||||
.pytest_cache/
|
||||
.mypy_cache/
|
||||
.ruff_cache/
|
||||
47
mcp/README.md
Normal file
47
mcp/README.md
Normal file
@ -0,0 +1,47 @@
|
||||
# todo backlog MCP server
|
||||
|
||||
A small, standalone, **read-only** [MCP](https://modelcontextprotocol.io)
|
||||
server that lets Claude Code (and its subagents) query the todo app's backlog.
|
||||
|
||||
The todo app is Dart/Flutter and stores notes in a `sqlite_crdt` database that
|
||||
must never be touched from outside the app (CRDT writes corrupt cross-device
|
||||
merge). Instead, this server reads the app's exported `BACKLOG.md` — the
|
||||
canonical NotesMarkdown snapshot written by **Settings → "Export notes"**.
|
||||
|
||||
## Staleness note
|
||||
|
||||
`BACKLOG.md` is a *manual* export, so it can be out of date. Every tool result
|
||||
includes the file's modification time (`mtime`) so you can judge freshness. If
|
||||
the file is missing, tools return `backlog_found: false` with a message telling
|
||||
you to export it from the app — they never crash.
|
||||
|
||||
## Tools (all read-only)
|
||||
|
||||
- `list_backlog(priority: str | None = None, status: str | None = None, limit: int = 100)`
|
||||
— notes, optionally filtered by `priority` (`low`/`medium`/`high`) and/or
|
||||
`status` (`todo`/`inProgress`/`done`/`abandoned`), plus `total`, `returned`,
|
||||
and `mtime`.
|
||||
- `get_note(note_id: str)` — one note by id, or `found: false`.
|
||||
- `backlog_stats()` — counts `by_status` and `by_priority`, `total`, and `mtime`.
|
||||
|
||||
Nothing here writes to the database or to `BACKLOG.md`.
|
||||
|
||||
## Setup
|
||||
|
||||
```bash
|
||||
mcp/scripts/setup_mcp.sh
|
||||
```
|
||||
|
||||
This creates `~/.venvs/todo-mcp`, installs this package (which pulls in the
|
||||
`mcp` SDK), and verifies the imports. The server is registered via the
|
||||
repo-root `.mcp.json`; restart Claude Code in this repo and approve the project
|
||||
MCP server prompt. Override the backlog location with `TODO_BACKLOG_PATH`
|
||||
(defaults to `~/todo/BACKLOG.md`).
|
||||
|
||||
## Development
|
||||
|
||||
```bash
|
||||
~/.venvs/todo-mcp/bin/pip install -e mcp pytest pytest-cov ruff mypy
|
||||
~/.venvs/todo-mcp/bin/python -m pytest mcp/todo_mcp/tests \
|
||||
--cov=todo_mcp --cov-branch --cov-report=term-missing --cov-fail-under=100
|
||||
```
|
||||
134
mcp/pyproject.toml
Normal file
134
mcp/pyproject.toml
Normal file
@ -0,0 +1,134 @@
|
||||
[project]
|
||||
name = "todo-mcp"
|
||||
version = "1.0.0"
|
||||
description = "Read-only MCP server over the todo app's exported BACKLOG.md"
|
||||
requires-python = ">=3.10"
|
||||
dependencies = [
|
||||
# The MCP SDK is the only runtime dependency: FastMCP powers the stdio
|
||||
# server. It lives in a dedicated venv (see scripts/setup_mcp.sh /
|
||||
# ../.mcp.json), kept entirely separate from the Dart/Flutter app.
|
||||
"mcp>=1.9",
|
||||
]
|
||||
|
||||
# Explicit package discovery so ``pip install -e .`` picks up the package and
|
||||
# not the scripts/ dir.
|
||||
[tool.setuptools.packages.find]
|
||||
include = ["todo_mcp*"]
|
||||
|
||||
# ============================================================================
|
||||
# RUFF - Fast Python linter and formatter
|
||||
# ============================================================================
|
||||
[tool.ruff]
|
||||
target-version = "py310"
|
||||
include = ["*.py", "**/*.py"]
|
||||
exclude = [".git", ".venv", "__pycache__", "build", "dist", ".eggs"]
|
||||
|
||||
[tool.ruff.lint]
|
||||
select = ["ALL"]
|
||||
ignore = [
|
||||
"D203", # conflicts with D211
|
||||
"D213", # conflicts with D212
|
||||
"COM812", # formatter handles this
|
||||
"ISC001", # formatter may create these
|
||||
"S603", # prone to false positives
|
||||
]
|
||||
fixable = ["ALL"]
|
||||
unfixable = []
|
||||
|
||||
[tool.ruff.lint.per-file-ignores]
|
||||
"**/tests/**/*.py" = ["ARG", "D", "PLC0415", "PLR2004", "S101", "SLF001"]
|
||||
"**/test_*.py" = ["ARG", "D", "PLC0415", "PLR2004", "S101", "SLF001"]
|
||||
|
||||
[tool.ruff.lint.pydocstyle]
|
||||
convention = "google"
|
||||
|
||||
[tool.ruff.lint.isort]
|
||||
force-single-line = false
|
||||
force-sort-within-sections = true
|
||||
known-first-party = ["todo_mcp"]
|
||||
|
||||
[tool.ruff.lint.flake8-quotes]
|
||||
docstring-quotes = "double"
|
||||
inline-quotes = "double"
|
||||
|
||||
[tool.ruff.lint.flake8-tidy-imports]
|
||||
ban-relative-imports = "all"
|
||||
|
||||
[tool.ruff.format]
|
||||
quote-style = "double"
|
||||
indent-style = "space"
|
||||
skip-magic-trailing-comma = false
|
||||
line-ending = "auto"
|
||||
docstring-code-format = true
|
||||
|
||||
# ============================================================================
|
||||
# MYPY - Static type checker
|
||||
# ============================================================================
|
||||
[tool.mypy]
|
||||
python_version = "3.10"
|
||||
strict = true
|
||||
warn_return_any = true
|
||||
warn_unused_configs = true
|
||||
disallow_untyped_defs = true
|
||||
disallow_incomplete_defs = true
|
||||
check_untyped_defs = true
|
||||
disallow_untyped_decorators = true
|
||||
no_implicit_optional = true
|
||||
warn_redundant_casts = true
|
||||
warn_unused_ignores = true
|
||||
warn_no_return = true
|
||||
warn_unreachable = true
|
||||
disallow_any_unimported = true
|
||||
disallow_any_explicit = false
|
||||
disallow_any_generics = true
|
||||
disallow_subclassing_any = true
|
||||
strict_equality = true
|
||||
extra_checks = true
|
||||
ignore_missing_imports = true
|
||||
show_error_codes = true
|
||||
color_output = true
|
||||
exclude = [".venv/"]
|
||||
|
||||
# ============================================================================
|
||||
# PYTEST
|
||||
# ============================================================================
|
||||
[tool.pytest.ini_options]
|
||||
testpaths = ["todo_mcp/tests"]
|
||||
python_files = ["test_*.py", "*_test.py"]
|
||||
python_classes = ["Test*"]
|
||||
python_functions = ["test_*"]
|
||||
addopts = [
|
||||
"-v",
|
||||
"--strict-markers",
|
||||
"--strict-config",
|
||||
"-ra",
|
||||
"--cov=todo_mcp",
|
||||
"--cov-branch",
|
||||
"--cov-report=term-missing",
|
||||
]
|
||||
filterwarnings = [
|
||||
"error",
|
||||
"ignore::DeprecationWarning",
|
||||
"default::pytest.PytestUnraisableExceptionWarning",
|
||||
]
|
||||
|
||||
# ============================================================================
|
||||
# COVERAGE
|
||||
# ============================================================================
|
||||
[tool.coverage.run]
|
||||
source = ["todo_mcp"]
|
||||
branch = true
|
||||
omit = ["*/__pycache__/*", "*/tests/*", "*/.venv/*"]
|
||||
|
||||
[tool.coverage.report]
|
||||
fail_under = 100
|
||||
show_missing = true
|
||||
skip_covered = false
|
||||
exclude_lines = [
|
||||
"pragma: no cover",
|
||||
"raise NotImplementedError",
|
||||
"raise AssertionError",
|
||||
"if TYPE_CHECKING:",
|
||||
'if __name__ == "__main__":',
|
||||
]
|
||||
partial_branches = ["pragma: no branch"]
|
||||
41
mcp/scripts/setup_mcp.sh
Executable file
41
mcp/scripts/setup_mcp.sh
Executable file
@ -0,0 +1,41 @@
|
||||
#!/bin/bash
|
||||
|
||||
# ============================================================================
|
||||
# Set up the dedicated virtualenv that hosts the todo backlog MCP server.
|
||||
# Claude Code spawns this interpreter (see the repo-root .mcp.json) to run
|
||||
# `python -m todo_mcp._mcp` over stdio.
|
||||
#
|
||||
# The MCP SDK (`mcp`) and the `todo_mcp` package live ONLY in this venv — the
|
||||
# todo app itself is Dart/Flutter and has no Python, so this is fully separate.
|
||||
# Both `mcp` and `todo_mcp` must be importable by this one interpreter or the
|
||||
# MCP server fails to start silently.
|
||||
#
|
||||
# Idempotent: safe to re-run to pick up dependency changes.
|
||||
# ============================================================================
|
||||
|
||||
set -euo pipefail
|
||||
|
||||
readonly SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
|
||||
readonly PKG_DIR="$(cd "${SCRIPT_DIR}/.." && pwd)"
|
||||
readonly VENV_DIR="${HOME}/.venvs/todo-mcp"
|
||||
|
||||
main() {
|
||||
echo "== Setting up MCP venv at ${VENV_DIR} =="
|
||||
|
||||
if [[ ! -d "${VENV_DIR}" ]]; then
|
||||
python3 -m venv "${VENV_DIR}"
|
||||
fi
|
||||
|
||||
"${VENV_DIR}/bin/pip" install --quiet --upgrade pip
|
||||
"${VENV_DIR}/bin/pip" install --quiet -e "${PKG_DIR}"
|
||||
|
||||
echo "== Verifying imports =="
|
||||
"${VENV_DIR}/bin/python" -c \
|
||||
"import mcp, todo_mcp; print('mcp + todo_mcp import OK')"
|
||||
|
||||
echo
|
||||
echo "Done. The server is registered via $(cd "${PKG_DIR}/.." && pwd)/.mcp.json"
|
||||
echo "Restart Claude Code in this repo and approve the project MCP server prompt."
|
||||
}
|
||||
|
||||
main "$@"
|
||||
17
mcp/todo_mcp/__init__.py
Normal file
17
mcp/todo_mcp/__init__.py
Normal file
@ -0,0 +1,17 @@
|
||||
"""Read-only MCP server package for the ``todo`` Flutter app's backlog.
|
||||
|
||||
The ``todo`` app is Dart/Flutter and stores notes in a ``sqlite_crdt`` database
|
||||
that must never be touched from here (CRDT writes corrupt cross-device merge).
|
||||
Instead, this package reads the app's exported ``BACKLOG.md`` file — the
|
||||
canonical, human-exportable "NotesMarkdown" snapshot — and exposes it to MCP
|
||||
clients (Claude Code and its subagents) as a handful of read-only tools.
|
||||
|
||||
See :mod:`todo_mcp._backlog` for the parser and :mod:`todo_mcp._mcp` for the
|
||||
FastMCP stdio server.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
__all__ = ["__version__"]
|
||||
|
||||
__version__ = "1.0.0"
|
||||
141
mcp/todo_mcp/_backlog.py
Normal file
141
mcp/todo_mcp/_backlog.py
Normal file
@ -0,0 +1,141 @@
|
||||
"""Parser for the ``todo`` app's exported ``BACKLOG.md`` (NotesMarkdown) file.
|
||||
|
||||
This is a faithful Python re-implementation of the Dart exporter/parser in
|
||||
``todo/lib/sync/notes_markdown.dart``. The exported document is valid Markdown:
|
||||
a one-line header (``<!-- todo-backlog v1 -->``) followed by, per note, an HTML
|
||||
comment metadata marker::
|
||||
|
||||
<!-- @note id="..." priority="medium" status="todo" created="..." updated="..." -->
|
||||
|
||||
whose body (everything up to the next marker, trimmed) is the note's Markdown.
|
||||
|
||||
The functions here are pure and side-effect free so they can be unit-tested in
|
||||
isolation; the MCP layer (:mod:`todo_mcp._mcp`) handles file I/O and staleness.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from dataclasses import dataclass
|
||||
import re
|
||||
import uuid
|
||||
|
||||
# Priority/status vocabularies mirror the Dart ``Priority``/``Status`` enums,
|
||||
# which are serialised by their ``.name`` (e.g. ``priority="medium"``). Unknown
|
||||
# or missing values fall back to the same defaults the Dart parser uses, so a
|
||||
# hand-edited file never yields an invalid note.
|
||||
_PRIORITIES = ("low", "medium", "high")
|
||||
_STATUSES = ("todo", "inProgress", "done", "abandoned")
|
||||
_DEFAULT_PRIORITY = "medium"
|
||||
_DEFAULT_STATUS = "todo"
|
||||
|
||||
# Matches a per-note metadata marker at the start of a line (multi-line mode).
|
||||
# The body is everything between one marker and the next (or end of file).
|
||||
_MARKER_PATTERN = re.compile(r"^<!--\s*@note\s+(.*?)\s*-->[ \t]*$", re.MULTILINE)
|
||||
|
||||
# Matches ``key="value"`` attribute pairs inside a marker.
|
||||
_ATTR_PATTERN = re.compile(r'(\w+)="([^"]*)"')
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class Note:
|
||||
"""A single parsed backlog note.
|
||||
|
||||
Attributes:
|
||||
id: Stable UUID from the marker, or a freshly minted one when the
|
||||
source marker had no (or a blank) ``id``.
|
||||
title: Human-readable title derived from the body's first line.
|
||||
body: The note's Markdown body, verbatim but trimmed.
|
||||
priority: One of ``low`` / ``medium`` / ``high``.
|
||||
status: One of ``todo`` / ``inProgress`` / ``done`` / ``abandoned``.
|
||||
"""
|
||||
|
||||
id: str
|
||||
title: str
|
||||
body: str
|
||||
priority: str
|
||||
status: str
|
||||
|
||||
|
||||
def _parse_attrs(raw: str) -> dict[str, str]:
|
||||
"""Extract ``key="value"`` pairs from a marker's attribute string.
|
||||
|
||||
Args:
|
||||
raw: The captured text between ``@note`` and the closing ``-->``.
|
||||
|
||||
Returns:
|
||||
A mapping of attribute name to its (possibly empty) string value.
|
||||
"""
|
||||
return {m.group(1): m.group(2) for m in _ATTR_PATTERN.finditer(raw)}
|
||||
|
||||
|
||||
def _normalize(value: str | None, allowed: tuple[str, ...], fallback: str) -> str:
|
||||
"""Return ``value`` if it is a recognised token, else ``fallback``.
|
||||
|
||||
Mirrors the Dart parser's tolerant enum resolution: a missing (``None``) or
|
||||
unknown value degrades to the enum's default rather than raising.
|
||||
|
||||
Args:
|
||||
value: The raw attribute value, or ``None`` if the marker omitted it.
|
||||
allowed: The permitted tokens for this field.
|
||||
fallback: The default token to use when ``value`` is unrecognised.
|
||||
|
||||
Returns:
|
||||
A valid token drawn from ``allowed``.
|
||||
"""
|
||||
return value if value in allowed else fallback
|
||||
|
||||
|
||||
def _derive_title(body: str) -> str:
|
||||
"""Derive a display title from a note's (already-trimmed) body.
|
||||
|
||||
Uses the first line, stripping any leading Markdown heading hashes so a
|
||||
``# Set up my own mail`` heading becomes ``Set up my own mail``. An empty
|
||||
body yields an empty title.
|
||||
|
||||
Args:
|
||||
body: The trimmed Markdown body of the note.
|
||||
|
||||
Returns:
|
||||
The derived title, or an empty string for an empty body.
|
||||
"""
|
||||
if not body:
|
||||
return ""
|
||||
return body.splitlines()[0].lstrip("#").strip()
|
||||
|
||||
|
||||
def parse_backlog(text: str) -> list[Note]:
|
||||
"""Parse an exported ``BACKLOG.md`` document into typed notes.
|
||||
|
||||
Tolerant by design (matching the Dart parser): a missing/blank ``id`` is
|
||||
replaced with a fresh UUID, and unknown/missing ``priority``/``status``
|
||||
values fall back to their defaults, so a partially hand-edited file never
|
||||
raises. The header line and any content before the first marker are ignored.
|
||||
|
||||
Args:
|
||||
text: The full contents of an exported NotesMarkdown document.
|
||||
|
||||
Returns:
|
||||
The notes in document order (empty if there are no ``@note`` markers).
|
||||
"""
|
||||
markers = list(_MARKER_PATTERN.finditer(text))
|
||||
notes: list[Note] = []
|
||||
for i, marker in enumerate(markers):
|
||||
attrs = _parse_attrs(marker.group(1))
|
||||
body_start = marker.end()
|
||||
body_end = markers[i + 1].start() if i + 1 < len(markers) else len(text)
|
||||
body = text[body_start:body_end].strip()
|
||||
|
||||
raw_id = attrs.get("id", "")
|
||||
note_id = raw_id or str(uuid.uuid4())
|
||||
notes.append(
|
||||
Note(
|
||||
id=note_id,
|
||||
title=_derive_title(body),
|
||||
body=body,
|
||||
priority=_normalize(
|
||||
attrs.get("priority"), _PRIORITIES, _DEFAULT_PRIORITY
|
||||
),
|
||||
status=_normalize(attrs.get("status"), _STATUSES, _DEFAULT_STATUS),
|
||||
)
|
||||
)
|
||||
return notes
|
||||
212
mcp/todo_mcp/_mcp.py
Normal file
212
mcp/todo_mcp/_mcp.py
Normal file
@ -0,0 +1,212 @@
|
||||
"""Read-only MCP server exposing the ``todo`` app's exported backlog.
|
||||
|
||||
Run via the dedicated venv that has ``mcp`` and ``todo_mcp`` installed::
|
||||
|
||||
~/.venvs/todo-mcp/bin/python -m todo_mcp._mcp
|
||||
|
||||
(see ``mcp/scripts/setup_mcp.sh`` and the repo-root ``.mcp.json``).
|
||||
|
||||
Safety invariants (do not break when adding tools):
|
||||
* **READ-ONLY.** Every tool only *reads* ``BACKLOG.md``. Nothing here writes
|
||||
to the file, and nothing ever touches the app's ``sqlite_crdt`` database
|
||||
(CRDT writes corrupt cross-device merge). There is no write tool.
|
||||
* **stdout is the JSON-RPC channel.** This module must never write to stdout;
|
||||
all logging is routed to STDERR below.
|
||||
* **No secret ever leaves.** The backlog holds only user-authored note text;
|
||||
there are no credentials to expose, and no config/token is ever read.
|
||||
|
||||
Because ``BACKLOG.md`` is a *manual* export (the app writes it from Settings →
|
||||
"Export notes"), it may be stale. Every result therefore includes the file's
|
||||
modification time (``mtime``) so a client can judge freshness, and a missing
|
||||
file yields a graceful "please export it" message rather than an error.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from datetime import datetime, timezone
|
||||
import logging
|
||||
import os
|
||||
from pathlib import Path
|
||||
import sys
|
||||
from typing import Any
|
||||
|
||||
from mcp.server.fastmcp import FastMCP
|
||||
|
||||
from todo_mcp._backlog import Note, parse_backlog
|
||||
|
||||
# Log to STDERR only — STDOUT carries the MCP JSON-RPC protocol frames, so a
|
||||
# single stray stdout write would corrupt the stream and kill the session.
|
||||
logging.basicConfig(
|
||||
level=logging.INFO,
|
||||
stream=sys.stderr,
|
||||
format="%(asctime)s [%(levelname)s] todo-mcp: %(message)s",
|
||||
)
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
mcp = FastMCP("todo")
|
||||
|
||||
_DEFAULT_LIST_LIMIT = 100
|
||||
|
||||
|
||||
def _backlog_path() -> Path:
|
||||
"""Resolve the ``BACKLOG.md`` path (env override, else the default).
|
||||
|
||||
Returns:
|
||||
``$TODO_BACKLOG_PATH`` when set, otherwise ``~/todo/BACKLOG.md``.
|
||||
"""
|
||||
override = os.environ.get("TODO_BACKLOG_PATH")
|
||||
if override:
|
||||
return Path(override)
|
||||
return Path.home() / "todo" / "BACKLOG.md"
|
||||
|
||||
|
||||
def _missing_message() -> str:
|
||||
"""Return the user-facing hint shown when the backlog file is absent."""
|
||||
return (
|
||||
f"BACKLOG.md not found at {_backlog_path()}. Export it from the todo "
|
||||
"app (Settings -> Export notes), then try again."
|
||||
)
|
||||
|
||||
|
||||
def _load_backlog() -> tuple[bool, list[Note], str | None]:
|
||||
"""Load and parse the backlog, reporting existence and file mtime.
|
||||
|
||||
Returns:
|
||||
A ``(found, notes, mtime)`` triple. When the file is absent ``found``
|
||||
is ``False`` and ``notes``/``mtime`` are empty; otherwise ``mtime`` is
|
||||
the file's UTC modification time as an ISO-8601 string.
|
||||
"""
|
||||
path = _backlog_path()
|
||||
try:
|
||||
stat = path.stat()
|
||||
except OSError:
|
||||
return False, [], None
|
||||
mtime = datetime.fromtimestamp(stat.st_mtime, tz=timezone.utc).isoformat()
|
||||
notes = parse_backlog(path.read_text(encoding="utf-8"))
|
||||
return True, notes, mtime
|
||||
|
||||
|
||||
def _note_to_dict(note: Note) -> dict[str, str]:
|
||||
"""Project a :class:`Note` into a JSON-serialisable dict for a tool result.
|
||||
|
||||
Args:
|
||||
note: The parsed note to serialise.
|
||||
|
||||
Returns:
|
||||
The note's fields as a plain string-keyed dict.
|
||||
"""
|
||||
return {
|
||||
"id": note.id,
|
||||
"title": note.title,
|
||||
"priority": note.priority,
|
||||
"status": note.status,
|
||||
"body": note.body,
|
||||
}
|
||||
|
||||
|
||||
@mcp.tool()
|
||||
def list_backlog(
|
||||
priority: str | None = None,
|
||||
status: str | None = None,
|
||||
limit: int = _DEFAULT_LIST_LIMIT,
|
||||
) -> dict[str, Any]:
|
||||
"""List backlog notes, optionally filtered by priority and/or status.
|
||||
|
||||
Args:
|
||||
priority: If given, keep only notes with this priority
|
||||
(``low`` / ``medium`` / ``high``).
|
||||
status: If given, keep only notes with this status
|
||||
(``todo`` / ``inProgress`` / ``done`` / ``abandoned``).
|
||||
limit: Maximum number of notes to return (non-positive returns none).
|
||||
|
||||
Returns:
|
||||
A dict with the matching ``notes`` (capped at ``limit``), the total
|
||||
match count, the number returned, and the backlog file ``mtime``. If
|
||||
the file is missing, ``backlog_found`` is ``False`` and a ``note`` field
|
||||
explains how to export it.
|
||||
"""
|
||||
found, notes, mtime = _load_backlog()
|
||||
if not found:
|
||||
return {"backlog_found": False, "note": _missing_message()}
|
||||
matches = [
|
||||
n
|
||||
for n in notes
|
||||
if (priority is None or n.priority == priority)
|
||||
and (status is None or n.status == status)
|
||||
]
|
||||
capped = matches[: max(0, limit)]
|
||||
return {
|
||||
"backlog_found": True,
|
||||
"mtime": mtime,
|
||||
"total": len(matches),
|
||||
"returned": len(capped),
|
||||
"notes": [_note_to_dict(n) for n in capped],
|
||||
}
|
||||
|
||||
|
||||
@mcp.tool()
|
||||
def get_note(note_id: str) -> dict[str, Any]:
|
||||
"""Fetch a single backlog note by its id.
|
||||
|
||||
Args:
|
||||
note_id: The note's UUID (the ``id`` from its marker).
|
||||
|
||||
Returns:
|
||||
A dict with ``found`` and, on a hit, the ``note`` plus the backlog
|
||||
``mtime``. If the file is missing, ``backlog_found`` is ``False`` and a
|
||||
``note`` field explains how to export it.
|
||||
"""
|
||||
found, notes, mtime = _load_backlog()
|
||||
if not found:
|
||||
return {"backlog_found": False, "found": False, "note": _missing_message()}
|
||||
for note in notes:
|
||||
if note.id == note_id:
|
||||
return {
|
||||
"backlog_found": True,
|
||||
"found": True,
|
||||
"mtime": mtime,
|
||||
"note": _note_to_dict(note),
|
||||
}
|
||||
return {
|
||||
"backlog_found": True,
|
||||
"found": False,
|
||||
"mtime": mtime,
|
||||
"note_id": note_id,
|
||||
}
|
||||
|
||||
|
||||
@mcp.tool()
|
||||
def backlog_stats() -> dict[str, Any]:
|
||||
"""Summarise the backlog: note counts by status and by priority.
|
||||
|
||||
Returns:
|
||||
A dict with the ``total`` note count, ``by_status`` and ``by_priority``
|
||||
count maps, and the backlog ``mtime``. If the file is missing,
|
||||
``backlog_found`` is ``False`` and a ``note`` field explains how to
|
||||
export it.
|
||||
"""
|
||||
found, notes, mtime = _load_backlog()
|
||||
if not found:
|
||||
return {"backlog_found": False, "note": _missing_message()}
|
||||
by_status: dict[str, int] = {}
|
||||
by_priority: dict[str, int] = {}
|
||||
for note in notes:
|
||||
by_status[note.status] = by_status.get(note.status, 0) + 1
|
||||
by_priority[note.priority] = by_priority.get(note.priority, 0) + 1
|
||||
return {
|
||||
"backlog_found": True,
|
||||
"mtime": mtime,
|
||||
"total": len(notes),
|
||||
"by_status": by_status,
|
||||
"by_priority": by_priority,
|
||||
}
|
||||
|
||||
|
||||
def main() -> None:
|
||||
"""Run the MCP server over stdio (STDOUT = JSON-RPC, STDERR = logs)."""
|
||||
logger.info("Starting todo MCP server (python=%s)", sys.executable)
|
||||
mcp.run() # pragma: no cover
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
1
mcp/todo_mcp/tests/__init__.py
Normal file
1
mcp/todo_mcp/tests/__init__.py
Normal file
@ -0,0 +1 @@
|
||||
"""Test package for todo_mcp."""
|
||||
96
mcp/todo_mcp/tests/test_backlog.py
Normal file
96
mcp/todo_mcp/tests/test_backlog.py
Normal file
@ -0,0 +1,96 @@
|
||||
"""Tests for the NotesMarkdown backlog parser in ``_backlog``."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import uuid
|
||||
|
||||
from todo_mcp._backlog import Note, parse_backlog
|
||||
|
||||
HEADER = "<!-- todo-backlog v1 -->\n\n"
|
||||
|
||||
|
||||
def _marker(**attrs: str) -> str:
|
||||
pairs = " ".join(f'{k}="{v}"' for k, v in attrs.items())
|
||||
return f"<!-- @note {pairs} -->"
|
||||
|
||||
|
||||
class TestParseBacklog:
|
||||
def test_empty_document_has_no_notes(self) -> None:
|
||||
assert parse_backlog("") == []
|
||||
|
||||
def test_header_only_has_no_notes(self) -> None:
|
||||
assert parse_backlog(HEADER) == []
|
||||
|
||||
def test_parses_a_single_note_with_heading_title(self) -> None:
|
||||
doc = (
|
||||
HEADER
|
||||
+ _marker(
|
||||
id="abc",
|
||||
priority="high",
|
||||
status="inProgress",
|
||||
)
|
||||
+ "\n# Set up my own mail\n\nbody text here\n"
|
||||
)
|
||||
notes = parse_backlog(doc)
|
||||
assert len(notes) == 1
|
||||
note = notes[0]
|
||||
assert note == Note(
|
||||
id="abc",
|
||||
title="Set up my own mail",
|
||||
body="# Set up my own mail\n\nbody text here",
|
||||
priority="high",
|
||||
status="inProgress",
|
||||
)
|
||||
|
||||
def test_plain_first_line_becomes_title_verbatim(self) -> None:
|
||||
doc = _marker(id="x", priority="low", status="todo") + "\nInfakt api + mcp\n"
|
||||
(note,) = parse_backlog(doc)
|
||||
assert note.title == "Infakt api + mcp"
|
||||
|
||||
def test_multiple_notes_split_on_markers(self) -> None:
|
||||
doc = (
|
||||
HEADER
|
||||
+ _marker(id="1", priority="low", status="todo")
|
||||
+ "\nfirst\n\n"
|
||||
+ _marker(id="2", priority="medium", status="done")
|
||||
+ "\nsecond\n"
|
||||
)
|
||||
notes = parse_backlog(doc)
|
||||
assert [n.id for n in notes] == ["1", "2"]
|
||||
assert notes[0].body == "first"
|
||||
assert notes[1].body == "second"
|
||||
|
||||
def test_missing_id_gets_fresh_uuid(self) -> None:
|
||||
doc = _marker(priority="low", status="todo") + "\nno id here\n"
|
||||
(note,) = parse_backlog(doc)
|
||||
# A valid UUID string was minted for the absent id.
|
||||
assert uuid.UUID(note.id)
|
||||
|
||||
def test_blank_id_gets_fresh_uuid(self) -> None:
|
||||
doc = _marker(id="", priority="low", status="todo") + "\nblank id\n"
|
||||
(note,) = parse_backlog(doc)
|
||||
assert uuid.UUID(note.id)
|
||||
|
||||
def test_unknown_priority_and_status_fall_back_to_defaults(self) -> None:
|
||||
doc = _marker(id="z", priority="urgent", status="waiting") + "\nx\n"
|
||||
(note,) = parse_backlog(doc)
|
||||
assert note.priority == "medium"
|
||||
assert note.status == "todo"
|
||||
|
||||
def test_absent_priority_and_status_fall_back_to_defaults(self) -> None:
|
||||
doc = _marker(id="z") + "\nx\n"
|
||||
(note,) = parse_backlog(doc)
|
||||
assert note.priority == "medium"
|
||||
assert note.status == "todo"
|
||||
|
||||
def test_empty_body_yields_empty_title(self) -> None:
|
||||
# Two adjacent markers => the first note's body is empty.
|
||||
doc = (
|
||||
_marker(id="1", priority="low", status="todo")
|
||||
+ "\n"
|
||||
+ _marker(id="2", priority="low", status="todo")
|
||||
+ "\nsecond\n"
|
||||
)
|
||||
notes = parse_backlog(doc)
|
||||
assert notes[0].body == ""
|
||||
assert notes[0].title == ""
|
||||
129
mcp/todo_mcp/tests/test_mcp.py
Normal file
129
mcp/todo_mcp/tests/test_mcp.py
Normal file
@ -0,0 +1,129 @@
|
||||
"""Tests for the read-only MCP tools in ``_mcp``.
|
||||
|
||||
Tools are driven directly (``@mcp.tool()`` leaves the function callable). File
|
||||
I/O is exercised for real against a temp ``BACKLOG.md`` pointed at via the
|
||||
``TODO_BACKLOG_PATH`` env var, so ``_backlog_path``'s override branch runs on
|
||||
every call; one test covers the default-path branch explicitly.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
|
||||
from todo_mcp import _mcp
|
||||
|
||||
_DOC = """<!-- todo-backlog v1 -->
|
||||
|
||||
<!-- @note id="n1" priority="high" status="todo" -->
|
||||
# First task
|
||||
alpha
|
||||
|
||||
<!-- @note id="n2" priority="low" status="done" -->
|
||||
# Second task
|
||||
beta
|
||||
|
||||
<!-- @note id="n3" priority="high" status="done" -->
|
||||
# Third task
|
||||
gamma
|
||||
"""
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def backlog(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> Path:
|
||||
path = tmp_path / "BACKLOG.md"
|
||||
path.write_text(_DOC, encoding="utf-8")
|
||||
monkeypatch.setenv("TODO_BACKLOG_PATH", str(path))
|
||||
return path
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def missing(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> Path:
|
||||
path = tmp_path / "absent.md"
|
||||
monkeypatch.setenv("TODO_BACKLOG_PATH", str(path))
|
||||
return path
|
||||
|
||||
|
||||
class TestBacklogPath:
|
||||
def test_default_path_when_env_unset(self, monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
monkeypatch.delenv("TODO_BACKLOG_PATH", raising=False)
|
||||
assert _mcp._backlog_path() == Path.home() / "todo" / "BACKLOG.md"
|
||||
|
||||
|
||||
class TestListBacklog:
|
||||
def test_lists_all_notes_with_mtime(self, backlog: Path) -> None:
|
||||
out = _mcp.list_backlog()
|
||||
assert out["backlog_found"] is True
|
||||
assert out["total"] == 3
|
||||
assert out["returned"] == 3
|
||||
assert out["mtime"] is not None
|
||||
assert [n["id"] for n in out["notes"]] == ["n1", "n2", "n3"]
|
||||
|
||||
def test_filter_by_priority(self, backlog: Path) -> None:
|
||||
out = _mcp.list_backlog(priority="high")
|
||||
assert [n["id"] for n in out["notes"]] == ["n1", "n3"]
|
||||
|
||||
def test_filter_by_status(self, backlog: Path) -> None:
|
||||
out = _mcp.list_backlog(status="done")
|
||||
assert [n["id"] for n in out["notes"]] == ["n2", "n3"]
|
||||
|
||||
def test_filter_by_priority_and_status(self, backlog: Path) -> None:
|
||||
out = _mcp.list_backlog(priority="high", status="done")
|
||||
assert [n["id"] for n in out["notes"]] == ["n3"]
|
||||
|
||||
def test_limit_caps_results(self, backlog: Path) -> None:
|
||||
out = _mcp.list_backlog(limit=1)
|
||||
assert out["total"] == 3
|
||||
assert out["returned"] == 1
|
||||
|
||||
def test_negative_limit_returns_none(self, backlog: Path) -> None:
|
||||
out = _mcp.list_backlog(limit=-3)
|
||||
assert out["returned"] == 0
|
||||
assert out["notes"] == []
|
||||
|
||||
def test_missing_file_is_graceful(self, missing: Path) -> None:
|
||||
out = _mcp.list_backlog()
|
||||
assert out["backlog_found"] is False
|
||||
assert "Export it" in out["note"]
|
||||
|
||||
|
||||
class TestGetNote:
|
||||
def test_found(self, backlog: Path) -> None:
|
||||
out = _mcp.get_note("n2")
|
||||
assert out["found"] is True
|
||||
assert out["note"]["title"] == "Second task"
|
||||
assert out["mtime"] is not None
|
||||
|
||||
def test_not_found_in_present_file(self, backlog: Path) -> None:
|
||||
out = _mcp.get_note("nope")
|
||||
assert out["backlog_found"] is True
|
||||
assert out["found"] is False
|
||||
assert out["note_id"] == "nope"
|
||||
|
||||
def test_missing_file_is_graceful(self, missing: Path) -> None:
|
||||
out = _mcp.get_note("n1")
|
||||
assert out["backlog_found"] is False
|
||||
assert out["found"] is False
|
||||
assert "Export it" in out["note"]
|
||||
|
||||
|
||||
class TestBacklogStats:
|
||||
def test_counts_by_status_and_priority(self, backlog: Path) -> None:
|
||||
out = _mcp.backlog_stats()
|
||||
assert out["total"] == 3
|
||||
assert out["by_status"] == {"todo": 1, "done": 2}
|
||||
assert out["by_priority"] == {"high": 2, "low": 1}
|
||||
assert out["mtime"] is not None
|
||||
|
||||
def test_missing_file_is_graceful(self, missing: Path) -> None:
|
||||
out = _mcp.backlog_stats()
|
||||
assert out["backlog_found"] is False
|
||||
assert "Export it" in out["note"]
|
||||
|
||||
|
||||
def test_main_runs_stdio_server(monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
calls: list[tuple[()]] = []
|
||||
monkeypatch.setattr(_mcp.mcp, "run", lambda: calls.append(()))
|
||||
_mcp.main()
|
||||
assert calls == [()]
|
||||
Loading…
Reference in New Issue
Block a user