From f0a3eaeec5ad0968e9e0de9afc9d4e4061862aa0 Mon Sep 17 00:00:00 2001 From: joungmin Date: Sat, 28 Feb 2026 07:42:31 +0900 Subject: [PATCH] feat: initial implementation of anki-mcp MCP server Implements a FastMCP server that wraps AnkiConnect (localhost:8765) to expose Anki flashcard management as Claude MCP tools. Tools provided: - list_decks: list all deck names - create_deck: create nested decks (:: separator) - add_card: add Basic flashcard with optional tags - get_due_cards: fetch today's due cards with optional deck filter - update_card: edit front/back of existing note - update_deck: rename deck (move-cards approach) Co-Authored-By: Claude Sonnet 4.6 --- .env.example | 1 + .gitignore | 10 +++ pyproject.toml | 21 ++++++ requirements.txt | 3 + src/__init__.py | 0 src/anki.py | 177 +++++++++++++++++++++++++++++++++++++++++++++++ src/server.py | 106 ++++++++++++++++++++++++++++ 7 files changed, 318 insertions(+) create mode 100644 .env.example create mode 100644 .gitignore create mode 100644 pyproject.toml create mode 100644 requirements.txt create mode 100644 src/__init__.py create mode 100644 src/anki.py create mode 100644 src/server.py diff --git a/.env.example b/.env.example new file mode 100644 index 0000000..8daf2f5 --- /dev/null +++ b/.env.example @@ -0,0 +1 @@ +ANKICONNECT_URL=http://localhost:8765 diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..68461c5 --- /dev/null +++ b/.gitignore @@ -0,0 +1,10 @@ +.env +.venv/ +__pycache__/ +*.pyc +*.pyo +.pytest_cache/ +*.egg-info/ +dist/ +build/ +.DS_Store diff --git a/pyproject.toml b/pyproject.toml new file mode 100644 index 0000000..35e6012 --- /dev/null +++ b/pyproject.toml @@ -0,0 +1,21 @@ +[build-system] +requires = ["setuptools>=68"] +build-backend = "setuptools.backends.legacy:build" + +[project] +name = "anki-mcp" +version = "0.1.0" +description = "MCP server for Anki flashcard management via AnkiConnect" +requires-python = ">=3.11" +dependencies = [ + "mcp[cli]", + "httpx", + "python-dotenv", +] + +[project.scripts] +anki-mcp = "src.server:main" + +[tool.setuptools.packages.find] +where = ["."] +include = ["src*"] diff --git a/requirements.txt b/requirements.txt new file mode 100644 index 0000000..8fb802a --- /dev/null +++ b/requirements.txt @@ -0,0 +1,3 @@ +mcp[cli] +httpx +python-dotenv diff --git a/src/__init__.py b/src/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/src/anki.py b/src/anki.py new file mode 100644 index 0000000..a116668 --- /dev/null +++ b/src/anki.py @@ -0,0 +1,177 @@ +"""AnkiConnect HTTP client. + +All calls use a single POST pattern to http://localhost:8765: + {"action": "actionName", "version": 6, "params": {...}} +""" + +import os +from typing import Any + +import httpx + +ANKICONNECT_URL = os.getenv("ANKICONNECT_URL", "http://localhost:8765") +_VERSION = 6 + + +def invoke(action: str, **params: Any) -> Any: + """Send a request to AnkiConnect and return the result. + + Args: + action: AnkiConnect action name. + **params: Action parameters. + + Returns: + The result field from the AnkiConnect response. + + Raises: + ConnectionError: If Anki is not running or AnkiConnect is unreachable. + ValueError: If AnkiConnect returns an error in the response. + """ + payload: dict[str, Any] = {"action": action, "version": _VERSION} + if params: + payload["params"] = params + + try: + response = httpx.post(ANKICONNECT_URL, json=payload, timeout=10.0) + response.raise_for_status() + except httpx.ConnectError as exc: + raise ConnectionError( + "Cannot connect to AnkiConnect. Make sure Anki is running with the " + "AnkiConnect add-on installed (code: 2055492159)." + ) from exc + except httpx.HTTPStatusError as exc: + raise ConnectionError(f"AnkiConnect returned HTTP {exc.response.status_code}") from exc + + data = response.json() + if data.get("error") is not None: + raise ValueError(f"AnkiConnect error: {data['error']}") + return data["result"] + + +def list_decks() -> list[str]: + """Return all deck names sorted alphabetically. + + Returns: + List of deck name strings. + """ + result: list[str] = invoke("deckNames") + return sorted(result) + + +def create_deck(deck_name: str) -> int: + """Create a new deck (or return existing deck ID). + + Args: + deck_name: Name of the deck to create. Use '::' for nested decks. + + Returns: + The deck ID as an integer. + """ + deck_id: int = invoke("createDeck", deck=deck_name) + return deck_id + + +def add_note( + deck_name: str, + front: str, + back: str, + tags: list[str] | None = None, +) -> int: + """Add a Basic card to a deck. + + Args: + deck_name: Target deck name. + front: Front (question) side of the card. + back: Back (answer) side of the card. + tags: Optional list of tags to attach. + + Returns: + The new note ID. + """ + note: dict[str, Any] = { + "deckName": deck_name, + "modelName": "Basic", + "fields": { + "Front": front, + "Back": back, + }, + "tags": tags or [], + "options": { + "allowDuplicate": False, + }, + } + note_id: int = invoke("addNote", note=note) + return note_id + + +def find_due_cards(deck_name: str = "", limit: int = 20) -> list[dict[str, Any]]: + """Find today's due cards, optionally filtered by deck. + + Args: + deck_name: Deck to filter by. Empty string means all decks. + limit: Maximum number of cards to return. + + Returns: + List of card info dicts with keys: cardId, noteId, deckName, front, back, due. + """ + query = "is:due" + if deck_name: + query = f'deck:"{deck_name}" {query}' + + card_ids: list[int] = invoke("findCards", query=query) + if not card_ids: + return [] + + # Respect limit before fetching full card info to reduce payload size + card_ids = card_ids[:limit] + cards_info: list[dict[str, Any]] = invoke("cardsInfo", cards=card_ids) + + result = [] + for card in cards_info: + result.append( + { + "cardId": card.get("cardId"), + "noteId": card.get("note"), + "deckName": card.get("deckName"), + "front": card.get("fields", {}).get("Front", {}).get("value", ""), + "back": card.get("fields", {}).get("Back", {}).get("value", ""), + "due": card.get("due"), + } + ) + return result + + +def update_note_fields(note_id: int, front: str = "", back: str = "") -> None: + """Update the Front and/or Back fields of an existing note. + + Args: + note_id: The note ID to update. + front: New front text. If empty, field is left unchanged. + back: New back text. If empty, field is left unchanged. + """ + fields: dict[str, str] = {} + if front: + fields["Front"] = front + if back: + fields["Back"] = back + if not fields: + return + + invoke("updateNoteFields", note={"id": note_id, "fields": fields}) + + +def rename_deck(deck_name: str, new_name: str) -> None: + """Rename a deck by creating the new deck and moving all cards into it. + + AnkiConnect does not expose a direct renameDeck action. + The old deck is left empty (Anki does not allow deleting decks without also + deleting cards, so we leave the empty shell to avoid data loss). + + Args: + deck_name: Current deck name. + new_name: New deck name. + """ + invoke("createDeck", deck=new_name) + card_ids: list[int] = invoke("findCards", query=f'deck:"{deck_name}"') + if card_ids: + invoke("changeDeck", cards=card_ids, deck=new_name) diff --git a/src/server.py b/src/server.py new file mode 100644 index 0000000..1dbedd5 --- /dev/null +++ b/src/server.py @@ -0,0 +1,106 @@ +"""Anki MCP server — manage Anki flashcards via AnkiConnect.""" + +import os +import sys + +# sys.path injection for `mcp dev` compatibility +_root = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) +if _root not in sys.path: + sys.path.insert(0, _root) + +from dotenv import load_dotenv + +load_dotenv(os.path.join(_root, ".env")) + +from mcp.server.fastmcp import FastMCP + +from src import anki + +mcp = FastMCP( + name="anki", + instructions=( + "Anki flashcard manager via AnkiConnect. " + "Add cards to decks, review due cards, manage categories. " + "Requires Anki to be running with the AnkiConnect add-on (code: 2055492159)." + ), +) + + +@mcp.tool() +def list_decks() -> list[str]: + """List all Anki deck names.""" + return anki.list_decks() + + +@mcp.tool() +def create_deck(deck_name: str) -> dict: + """Create a new Anki deck (or return existing deck ID). + + Args: + deck_name: Deck name. Use '::' separator for nested decks (e.g. 'Korean::Vocabulary'). + """ + deck_id = anki.create_deck(deck_name) + return {"deck_name": deck_name, "deck_id": deck_id} + + +@mcp.tool() +def add_card( + deck_name: str, + front: str, + back: str, + tags: list[str] = [], +) -> dict: + """Add a Basic flashcard to a deck. + + Args: + deck_name: Target deck name (must already exist or be created first). + front: Front (question) side of the card. + back: Back (answer) side of the card. + tags: Optional list of tags to attach to the card. + """ + note_id = anki.add_note(deck_name, front, back, tags) + return {"note_id": note_id, "deck_name": deck_name, "front": front, "back": back} + + +@mcp.tool() +def get_due_cards(deck_name: str = "", limit: int = 20) -> list[dict]: + """Get today's due cards for review. + + Args: + deck_name: Filter by deck name. Leave empty to get due cards from all decks. + limit: Maximum number of cards to return (default 20). + """ + return anki.find_due_cards(deck_name, limit) + + +@mcp.tool() +def update_card(note_id: int, front: str = "", back: str = "") -> dict: + """Edit an existing card's Front and/or Back content. + + Args: + note_id: The note ID to update (obtained from add_card or get_due_cards). + front: New front text. Leave empty to keep existing. + back: New back text. Leave empty to keep existing. + """ + anki.update_note_fields(note_id, front, back) + return {"note_id": note_id, "updated": {"front": bool(front), "back": bool(back)}} + + +@mcp.tool() +def update_deck(deck_name: str, new_name: str) -> dict: + """Rename an existing deck. + + Args: + deck_name: Current deck name. + new_name: New deck name. + """ + anki.rename_deck(deck_name, new_name) + return {"old_name": deck_name, "new_name": new_name} + + +def main() -> None: + mcp.run() + + +if __name__ == "__main__": + main()