MCP-AnkiConnect
by samefarrar
- mcp_ankiconnect
from typing import List, Optional, Literal, Dict, Union
import json
import random
import logging
from contextlib import asynccontextmanager
from mcp.server.fastmcp import FastMCP
from mcp_ankiconnect.ankiconnect_client import AnkiConnectClient
from mcp_ankiconnect.config import EXCLUDE_STRINGS, RATING_TO_EASE
from mcp_ankiconnect.server_prompts import flashcard_guidelines, claude_review_instructions
from pydantic import Field
logger = logging.getLogger(__name__)
logger.info("Initializing MCP-AnkiConnect server")
mcp = FastMCP("mcp-ankiconnect")
logger.debug("Created FastMCP instance")
@asynccontextmanager
async def get_anki_client():
client = AnkiConnectClient()
try:
yield client
finally:
await client.close()
mcp = FastMCP("mcp-ankiconnect")
async def get_cards_by_due_and_deck(deck: Optional[str] = None, day: Optional[int] = 0) -> List[int]:
async with get_anki_client() as anki:
decks = await anki.deck_names()
if deck and deck not in decks:
raise ValueError(f"Deck '{deck}' does not exist")
if day > 0:
prop = f"prop:due<{day+1}"
else:
prop = "prop:due<=0"
# Construct the search query
query = f"is:due -is:suspended {prop}"
if deck:
query += f' deck:{deck}'
# Get and return the due cards
return await anki.find_cards(query=query)
@mcp.tool()
async def num_cards_due_today(deck: Optional[str] = None) -> str:
"""Get the number of cards due today with an optional deck filter"""
anki = AnkiConnectClient()
card_ids = await get_cards_by_due_and_deck(deck, 0)
if deck:
return f"There are {len(card_ids)} cards due in deck '{deck}'"
else:
return f"There are {len(card_ids)} cards due across all decks"
@mcp.tool()
async def list_decks_and_notes() -> str:
"""Get all decks and note types with their fields"""
anki = AnkiConnectClient()
decks = await anki.deck_names()
decks = [deck for deck in decks if not any(exclude.lower() in deck.lower() for exclude in EXCLUDE_STRINGS)]
model_names = await anki.model_names()
note_types = []
for model in model_names:
if any(exclude.lower() in model.lower() for exclude in EXCLUDE_STRINGS):
continue
fields = await anki.model_field_names(model)
note_types.append({"name": model, "fields": fields})
result = f"""You have {len(decks)} decks: {', '.join(decks)}
Your note types are: {[note['name'] for note in note_types]}:
{chr(10).join([f"{note['name']}" + ': { ' + ', '.join([f'"{field}": "string"' for field in note['fields']]) + ' }' for note in note_types])}
"""
return result
@mcp.tool()
async def get_examples(
deck: Optional[str] = None,
limit: int = Field(default = 5, ge = 1),
sample: str = Field(
default = "random",
pattern="^(random|recent|most_reviewed|best_performance|mature|young)$"
))-> str:
"""Get example notes from Anki to guide your flashcard making. Limit the number of examples returned and provide a sampling technique:
- random: Randomly sample notes
- recent: Notes added in the last week
- most_reviewed: Notes with more than 10 reviews
- best_performance: Notes with less than 3 lapses
- mature: Notes with interval greater than 21 days
- young: Notes with interval less than 7 days
"""
anki = AnkiConnectClient()
query = "-is:suspended " + " ".join([f"-note:*{exclude_string}*" for exclude_string in EXCLUDE_STRINGS]) + " "
if deck:
query += f"deck:{deck} "
match sample:
case "recent":
query += "added:7" # Added in last week
case "most_reviewed":
query += "prop:reps>10 -is:learn" # Reviewed cards excluding learning
case "best_performance":
query += "prop:lapses<3 is:review" # Review cards with few lapses
case "mature":
query += "prop:ivl>=21 -is:learn" # Mature cards not in learning
case "young":
query += "is:review prop:ivl<=7 -is:learn" # Young review cards
case "random":
query += "prop:due<180"
note_ids = await anki.find_notes(query=query)
if len(note_ids) > limit and sample == "random":
note_ids = random.sample(note_ids, limit)
note_ids = note_ids[:limit]
if not note_ids:
return "No example notes found matching criteria"
notes = await anki.notes_info(note_ids)
examples = []
for note in notes:
example = {
"tags": note["tags"],
"modelName": note["modelName"],
"fields": {name: {k: v for k, v in value.items() if k != "order"} for name, value in note["fields"].items()}
}
examples.append(example)
result = flashcard_guidelines + "\n" + json.dumps(examples, indent=2)
return result
@mcp.tool()
async def fetch_due_cards_for_review(
deck: Optional[str] = None,
limit: int = 5,
today_only: bool = True,
) -> str:
"""Fetch cards that are due for learning and format them for review. Takes optional arguments:
- deck: str - Filter by specific deck.
- limit: int - Maximum number of cards to fetch (default 5). More than 5 is overwhelming for users.
- today_only: bool - If true, only fetch cards due today, else fetch cards up to 5 days ahead."""
anki = AnkiConnectClient()
days_to_review = 0 if today_only else 5
card_ids = await get_cards_by_due_and_deck(deck, days_to_review)
card_ids = card_ids[:limit]
cards = await anki.cards_info(card_ids=card_ids)
# Format the card information into a readable message
cards_info = []
for card in cards:
# Get question fields (where order != fieldOrder)
question_fields = [
f"<{name.lower()}>{field['value']}</{name.lower()}>"
for name, field in card['fields'].items()
if field['order'] == card['fieldOrder']
]
# Get answer fields (where order == fieldOrder)
answer_fields = [
f"<{name.lower()}>{field['value']}</{name.lower()}>"
for name, field in list(card['fields'].items())[:5]
if field['order'] != card['fieldOrder']
]
cards_info.append(
f"<card id=\"{card['cardId']}\">\n"
f" <question>{'; '.join(question_fields)}</question>\n"
f" <answer>{'; '.join(answer_fields)}</answer>\n"
f"</card>")
cards_text = "\n\n".join(cards_info)
if not cards_text:
cards_text = "No cards found to review"
examples_prompt = claude_review_instructions.replace("{{flashcards}}", cards_text)
return examples_prompt
@mcp.tool()
async def submit_reviews(reviews: List[Dict[Literal["card_id", "rating"], Union[int, Literal["wrong", "hard", "good", "easy"]]]]) -> str:
"""Submit multiple card reviews to Anki.
Args:
reviews: List of dictionaries containing:
- card_id (int): The ID of the card being reviewed
- rating (str): The rating to give the card, one of:
"wrong" - Card was incorrect (Again)
"hard" - Card was difficult (Hard)
"good" - Card was good (Good)
"easy" - Card was very easy (Easy)
"""
anki = AnkiConnectClient()
if not reviews:
raise ValueError("Arguments required for submitting reviews")
# Convert reviews to AnkiConnect format
answers = [
{"cardId": review["card_id"], "ease": RATING_TO_EASE[review["rating"]]}
for review in reviews
]
# Submit all reviews at once
results = await anki.answer_cards(answers)
# Generate response messages
messages = [
f"Card {review['card_id']} {'successfully' if success else 'failed to be'} marked as {review['rating']}"
for review, success in zip(reviews, results)
]
return "\n".join(messages)
@mcp.tool()
async def add_note(
deckName: str,
modelName: str,
fields: dict[str, str],
tags: List[str] = Field(default_factory = list)) -> str:
"""Add a flashcard to Anki. Ensure you have looked at examples before you do this, and that you have got approval from the user to add the flashcard.
Args:
deckName: str - The name of the deck to add the flashcard to.
modelName: str - The name of the note type to use.
fields: dict - The fields of the flashcard to add.
tags: List[str] - The tags to add to the flashcard."""
anki = AnkiConnectClient()
note = {
"deckName": deckName,
"modelName": modelName,
"fields": fields,
"tags": tags,
"options": {
"allowDuplicate": False,
"duplicateScope": "deck",
}
}
note_id = await anki.add_note(note)
return f"Successfully created note with ID: {note_id}"