MCP-AnkiConnect

import functools # Import functools import json import logging import random import re from collections.abc import AsyncGenerator from contextlib import asynccontextmanager from typing import Any, Literal from mcp.server.fastmcp import FastMCP from pydantic import Field # Use relative imports within the package from .ankiconnect_client import ( # Import custom exception AnkiConnectClient, AnkiConnectionError, ) from .config import ( # Import necessary configs ANKI_CONNECT_URL, EXCLUDE_STRINGS, MAX_FUTURE_DAYS, RATING_TO_EASE, ) from .server_prompts import claude_review_instructions, flashcard_guidelines logger = logging.getLogger(__name__) logger.info("Initializing MCP-AnkiConnect server") mcp = FastMCP("mcp-ankiconnect") logger.debug("Created FastMCP instance") # --- Context Manager for Client --- @asynccontextmanager async def get_anki_client() -> AsyncGenerator[ AnkiConnectClient, None ]: # Added type hint # Pass the configured URL to the client constructor client = AnkiConnectClient(base_url=ANKI_CONNECT_URL) try: yield client except Exception: # Log exceptions during client usage if needed logger.exception("Error occurred while using AnkiConnect client") raise # Re-raise the exception finally: logger.debug("Closing AnkiConnect client") await client.close() # --- Decorator for Connection Error Handling --- def handle_anki_connection_error(func): """Decorator to catch AnkiConnectionError and return a user-friendly message.""" @functools.wraps(func) async def wrapper(*args, **kwargs): try: # Call the original async tool function return await func(*args, **kwargs) except AnkiConnectionError as e: logger.error(f"Caught Anki connection error in tool '{func.__name__}': {e}") # Return the specific message intended for the LLM # Ensure the message clearly indicates it's an error for the LLM to handle return ( "SYSTEM_ERROR: Cannot connect to Anki. " "Please inform the user that they need to start their Anki application " "and ensure the AnkiConnect add-on is installed and enabled before proceeding. " f"Details: {e}" # Include the error detail from the exception ) except ValueError as e: # Catch Anki API errors (raised as ValueError from invoke) logger.error(f"Caught Anki API error in tool '{func.__name__}': {e}") return ( f"SYSTEM_ERROR: An error occurred communicating with Anki: {e}. " "Please inform the user about the error." ) except Exception as e: # Catch any other unexpected errors within the tool logger.exception( f"An unexpected error occurred in tool '{func.__name__}': {e}" ) # Provide a generic error message for the LLM return ( f"SYSTEM_ERROR: An unexpected error occurred while executing the Anki tool '{func.__name__}'. " f"Details: {e}" ) return wrapper # --- End Decorator --- # --- Helper Function (Refactored) --- # This helper now requires the client to be passed in, making it testable # and ensuring it runs within the context managed by the tool. async def _find_due_card_ids( client: AnkiConnectClient, deck: str | None = None, day: int | None = 0 ) -> list[int]: """Finds card IDs due on a specific day relative to today (0=today).""" if day < 0: raise ValueError("Day must be non-negative.") # Construct the search query # prop:due=0 means due today # prop:due=1 means due tomorrow (relative to review time) # prop:due<=N finds cards due today up to N days in the future. if day == 0: prop = "prop:due=0" # Due exactly today else: prop = f"prop:due<={day}" query = f"is:due -is:suspended {prop}" if deck: # Add deck to query, ensuring proper quoting for spaces query += f' "deck:{deck}"' logger.debug(f"Executing Anki card search query: {query}") card_ids = await client.find_cards(query=query) logger.info(f"Found {len(card_ids)} cards for query: {query}") return card_ids def _build_example_query(deck: str | None, sample: str) -> str: """Builds the Anki query string for finding example notes. Note: AnkiConnect's findNotes does not support sort: directives — those are a browser UI feature only. Ordering/selection is handled in Python after the results are returned. """ query_parts = ["-is:suspended"] query_parts.extend([f"-note:*{ex}*" for ex in EXCLUDE_STRINGS]) if deck: query_parts.append(f'"deck:{deck}"') match sample: case "recent": query_parts.append("added:7") case "most_reviewed": query_parts.append("prop:reps>10") case "best_performance": query_parts.append("prop:lapses<3 is:review") case "mature": query_parts.append("prop:ivl>=21 -is:learn") case "young": query_parts.append("is:review prop:ivl<=7 -is:learn") case "random": query_parts.append("is:review") return " ".join(query_parts) def _format_example_notes(notes_info: list[dict]) -> list[dict]: """Formats note information into simplified dictionaries for examples.""" examples = [] for note in notes_info: processed_fields = {} for name, field_data in note.get("fields", {}).items(): value = field_data.get("value", "") processed_value = value.replace("<pre><code>", "<code>").replace( "</code></pre>", "</code>" ) processed_fields[name] = processed_value example = { "modelName": note.get("modelName", "UnknownModel"), "fields": processed_fields, "tags": note.get("tags", []), } examples.append(example) return examples def _format_search_results(notes_info: list[dict]) -> list[dict]: """Formats note search results for LLM consumption. Includes note IDs to enable follow-up actions like editing or deletion. """ results = [] for note in notes_info: processed_fields = {} for name, field_data in note.get("fields", {}).items(): value = field_data.get("value", "") # Clean up code formatting for readability processed_value = value.replace("<pre><code>", "<code>").replace( "</code></pre>", "</code>" ) processed_fields[name] = processed_value result = { "noteId": note.get("noteId"), "modelName": note.get("modelName", "UnknownModel"), "fields": processed_fields, "tags": note.get("tags", []), } results.append(result) return results def _format_cards_for_llm(cards_info: list[dict]) -> str: """Formats card information into an XML-like string for the LLM.""" formatted_cards = [] for card in cards_info: card_id = card.get("cardId", "UNKNOWN_ID") fields = card.get("fields", {}) question_field_order = card.get("fieldOrder", 0) question_parts = [] answer_parts = [] sorted_field_items = sorted( fields.items(), key=lambda item: item[1].get("order", 0) ) for name, field_data in sorted_field_items: field_value = field_data.get("value", "") field_order = field_data.get("order", -1) tag_name = name.lower().replace(" ", "_") if field_order == question_field_order: question_parts.append(f"<{tag_name}>{field_value}</{tag_name}>") else: answer_parts.append(f"<{tag_name}>{field_value}</{tag_name}>") question_str = ( "".join(question_parts) if question_parts else "<error>Question field not found</error>" ) answer_str = ( " ".join(answer_parts) if answer_parts else "<error>Answer fields not found</error>" ) formatted_cards.append( f'<card id="{card_id}">\n' f" <question>{question_str}</question>\n" f" <answer>{answer_str}</answer>\n" f"</card>" ) return "\n\n".join(formatted_cards) def _process_field_content(content: str) -> str: """Processes field content for MathJax and code blocks before sending to Anki.""" if not isinstance(content, str): logger.warning( f"Field content is not a string (type: {type(content)}). Returning as-is." ) return content # Return non-strings unmodified # 1. MathJax: <math>...</math> -> \(...\) processed_value = content.replace("<math>", "\\(").replace("</math>", "\\)") # 2. Code Blocks: ```lang\n...\n``` -> <pre><code class="language-lang">...</code></pre> processed_value = re.sub( r"```(\w+)?\s*\n?(.*?)```", lambda m: f'<pre><code class="language-{m.group(1)}">{m.group(2)}</code></pre>' if m.group(1) else f"<pre><code>{m.group(2)}</code></pre>", processed_value, flags=re.DOTALL, ) # 3. Inline Code: `...` -> <code>...</code> processed_value = re.sub(r"`([^`]+)`", r"<code>\1</code>", processed_value) return processed_value # --- Tool Definitions --- @mcp.tool() @handle_anki_connection_error # Apply decorator async def num_cards_due_today(deck: str | None = None) -> str: """Get the number of cards due exactly today, with an optional deck filter.""" async with get_anki_client() as anki: # Use the helper, specifying day=0 for today card_ids = await _find_due_card_ids(anki, deck, day=0) count = len(card_ids) deck_msg = f" in deck '{deck}'" if deck else " across all decks" return f"There are {count} cards due today{deck_msg}." @mcp.tool() @handle_anki_connection_error # Apply decorator async def list_decks_and_notes() -> str: """Get all decks (excluding specified patterns) and note types with their fields.""" async with get_anki_client() as anki: all_decks = await anki.deck_names() # Filter decks based on EXCLUDE_STRINGS decks = [ d for d in all_decks if not any(ex.lower() in d.lower() for ex in EXCLUDE_STRINGS) ] logger.info(f"Filtered decks: {decks}") all_model_names = await anki.model_names() note_types = [] for model in all_model_names: if any(ex.lower() in model.lower() for ex in EXCLUDE_STRINGS): continue try: fields = await anki.model_field_names(model) note_types.append({"name": model, "fields": fields}) except Exception as e: logger.warning( f"Could not get fields for model '{model}': {e}. Skipping this model." ) # Format the output string deck_list_str = ( f"You have {len(decks)} filtered decks: {', '.join(decks)}" if decks else "No filtered decks found." ) note_type_list = [] if note_types: for note in note_types: # Format fields as "FieldName": "type" (assuming string for simplicity) field_str = ", ".join( [f'"{field}": "string"' for field in note["fields"]] ) note_type_list.append(f"- {note['name']}: {{ {field_str} }}") note_types_str = ( "Your filtered note types and their fields are:\n" + "\n".join(note_type_list) ) else: note_types_str = "No filtered note types found." return f"{deck_list_str}\n\n{note_types_str}" @mcp.tool() @handle_anki_connection_error # Apply decorator async def get_examples( deck: str | None = None, limit: int = Field(default=5, ge=1), sample: str = Field( default="random", description="Sampling technique: random, recent (added last 7d), most_reviewed (>10 reps), best_performance (<3 lapses), mature (ivl>=21d), young (ivl<=7d)", pattern="^(random|recent|most_reviewed|best_performance|mature|young)$", # Keep pattern for validation ), # Close Field() ) -> str: # Close parameters list """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 Args: deck: Optional[str] - Filter by specific deck (use exact name). limit: int - Maximum number of examples to return (default 5). sample: str - Sampling technique (random, recent, most_reviewed, best_performance, mature, young). """ async with get_anki_client() as anki: # Build the query using the helper function query = _build_example_query(deck, sample) logger.debug(f"Finding example notes with query: {query}") note_ids = await anki.find_notes(query=query) if not note_ids: return f"No example notes found matching criteria (Sample: {sample}, Deck: {deck or 'Any'})." # Apply sampling and limit if sample == "random" and len(note_ids) > limit: sampled_note_ids = random.sample(note_ids, limit) else: # For sorted queries, take the top results up to the limit sampled_note_ids = note_ids[:limit] if not sampled_note_ids: return f"No example notes found after sampling/limiting (Sample: {sample}, Deck: {deck or 'Any'})." logger.debug(f"Fetching info for note IDs: {sampled_note_ids}") notes_info = await anki.notes_info(sampled_note_ids) # Format notes using the helper function formatted_examples = _format_example_notes(notes_info) # Combine guidelines with the JSON examples # Use json.dumps for clean formatting examples_json = json.dumps(formatted_examples, indent=2, ensure_ascii=False) result = f"{flashcard_guidelines}\n\nHere are some examples based on your criteria:\n{examples_json}" return result @mcp.tool() @handle_anki_connection_error # Apply decorator async def fetch_due_cards_for_review( deck: str | None = None, limit: int = Field(default=5, ge=1, description="Max cards to fetch."), today_only: bool = Field( default=True, description="True=only today's cards, False=cards due up to MAX_FUTURE_DAYS ahead.", ), ) -> str: """Fetch cards due for review, formatted for an LLM to present. Args: deck: Optional[str] - Filter by specific deck name. limit: int - Maximum number of cards to fetch (default 5). today_only: bool - If true, only fetch cards due today. If false, fetch cards due up to MAX_FUTURE_DAYS ahead (currently {MAX_FUTURE_DAYS}). """ async with get_anki_client() as anki: # Determine the maximum relative day to check based on today_only flag # day=0 means today, day=MAX_FUTURE_DAYS means today up to MAX_FUTURE_DAYS from now max_day_to_check = 0 if today_only else MAX_FUTURE_DAYS # Use the helper function to find relevant card IDs card_ids = await _find_due_card_ids(anki, deck, day=max_day_to_check) # Limit the number of cards to fetch info for card_ids_to_fetch = card_ids[:limit] if not card_ids_to_fetch: deck_msg = f" in deck '{deck}'" if deck else "" when_msg = ( "today" if today_only else f"within the next {max_day_to_check} days" ) return f"No cards found due {when_msg}{deck_msg}." logger.debug(f"Fetching info for card IDs: {card_ids_to_fetch}") cards_info_list = await anki.cards_info(card_ids=card_ids_to_fetch) # Format cards using the helper function cards_text = _format_cards_for_llm(cards_info_list) # Inject the formatted cards into the review instructions prompt review_prompt = claude_review_instructions.replace("{{flashcards}}", cards_text) return review_prompt @mcp.tool() @handle_anki_connection_error # Apply decorator async def submit_reviews( reviews: list[ dict[ Literal["card_id", "rating"], int | Literal["wrong", "hard", "good", "easy"] ] ], ) -> str: """Submit multiple card reviews to Anki using ratings ('wrong', 'hard', 'good', 'easy'). Args: reviews: List of review dictionaries, each with: - card_id (int): The ID of the card reviewed. - rating (str): 'wrong', 'hard', 'good', or 'easy'. """ if not reviews: # Return a message instead of raising ValueError, handled by decorator now return "No reviews provided to submit." answers_to_submit = [] validation_errors = [] for review in reviews: card_id = review.get("card_id") rating = str(review.get("rating", "")).lower() # Ensure lowercase string if not isinstance(card_id, int): validation_errors.append( f"Invalid card_id '{card_id}' in review: {review}. Must be an integer." ) continue # Skip this invalid review ease = RATING_TO_EASE.get(rating) if ease is None: valid_ratings = list(RATING_TO_EASE.keys()) validation_errors.append( f"Invalid rating '{rating}' for card_id {card_id}. Must be one of: {valid_ratings}." ) continue # Skip this invalid review answers_to_submit.append({"cardId": card_id, "ease": ease}) if validation_errors: # Report validation errors back to the LLM/user errors_str = "\n".join(validation_errors) return f"SYSTEM_ERROR: Could not submit reviews due to validation errors:\n{errors_str}" if not answers_to_submit: return "No valid reviews found to submit after validation." async with get_anki_client() as anki: logger.info(f"Submitting {len(answers_to_submit)} reviews to Anki.") # The 'answerCards' action returns a list of booleans (or similar success indicators) # It might raise an error if the entire batch fails, handled by invoke/decorator results = await anki.answer_cards(answers=answers_to_submit) # Check if the result length matches the input length if len(results) != len(answers_to_submit): logger.warning( f"Anki response length mismatch: Expected {len(answers_to_submit)}, Got {len(results)}" ) # Handle potential mismatch - maybe return a generic success/fail message # For now, assume results correspond to input order if length matches # Generate response messages based on results (assuming True means success) messages = [] success_count = 0 fail_count = 0 for i, review in enumerate( reviews ): # Iterate original reviews to get card_id/rating card_id = review["card_id"] rating = review["rating"] # Check corresponding result if lengths match, otherwise assume failure? success = ( results[i] if i < len(results) else False ) # Default to False if mismatch if success: messages.append(f"Card {card_id}: Marked as '{rating}' successfully.") success_count += 1 else: messages.append(f"Card {card_id}: Failed to mark as '{rating}'.") fail_count += 1 summary = f"Review submission summary: {success_count} successful, {fail_count} failed." full_response = summary + "\n" + "\n".join(messages) logger.info(full_response) return full_response @mcp.tool() @handle_anki_connection_error # Apply decorator async def add_note( deckName: str, modelName: str, fields: dict[str, str], tags: list[str] | None = None, picture: list[dict[str, str | list[str]]] | None = None, ) -> 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. For code examples, use <code> tags to format your code. e.g. <code>def fibonacci(n): if n <= 1: return n return fibonacci(n-1) + fibonacci(n-2)</code> For MathJax, use the <math> tag to format your math equations. This will automatically render the math equations in Anki. # e.g. <math>\\frac{d}{dx}[3\\sin(5x)] = 15\\cos(5x)</math> To attach images to a card, use the picture parameter. Each picture object must have a filename and exactly one source (url, data, or path). The fields list specifies which card fields get the <img> tag inserted. ## How to attach images based on the source: **User provides a URL:** [{"url": "https://example.com/photo.jpg", "filename": "photo.jpg", "fields": ["Back"]}] **User provides a local file (e.g. screenshot, downloaded image):** [{"path": "/absolute/path/to/image.png", "filename": "image.png", "fields": ["Back"]}] **Base64-encoded data (for small images only):** [{"data": "iVBORw0KGgo...", "filename": "diagram.png", "fields": ["Back"]}] IMPORTANT: When a user shares an image file or screenshot, prefer using "path" with the absolute file path rather than trying to base64-encode the image contents. AnkiConnect reads the file directly from disk which is faster and more reliable. Args: deckName: str - The target deck name. modelName: str - The note type (model) name. fields: dict - Dictionary of field names and their string content. tags: List[str] - Optional list of tags. picture: List[dict] - Optional list of picture attachments. Each dict should have: - filename (str): Name for the stored image file. - url (str, optional): URL to download the image from. - path (str, optional): Absolute file path to a local image. Preferred for user-shared files. - data (str, optional): Base64-encoded image data. - fields (List[str]): Card field names where the <img> tag will be inserted. - skipHash (str, optional): MD5 hash to skip if file matches. """ # Process fields using the helper function processed_fields = { name: _process_field_content(value) for name, value in fields.items() } note_payload: dict[str, Any] = { "deckName": deckName, "modelName": modelName, "fields": processed_fields, # Use processed fields "tags": tags if tags is not None else [], "options": { "allowDuplicate": False, "duplicateScope": "deck", }, } if picture: note_payload["picture"] = picture async with get_anki_client() as anki: logger.info( f"Attempting to add note to deck '{deckName}' with model '{modelName}'." ) if picture: logger.info(f"Note includes {len(picture)} picture attachment(s).") logger.debug( f"Note Payload: {json.dumps(note_payload, indent=2)}" ) # Log the payload for debugging # Invoke addNote - errors (like duplicate, missing fields) will be caught # by the ValueError check in invoke or the decorator note_id = await anki.add_note(note=note_payload) # add_note returns the new note ID on success, or raises error/returns None/0 on failure if note_id: success_message = ( f"Successfully created note with ID: {note_id} in deck '{deckName}'." ) if picture: success_message += f" ({len(picture)} image(s) attached)" logger.info(success_message) return success_message else: fail_message = f"Failed to add note to deck '{deckName}'. AnkiConnect did not return a note ID or indicated failure." logger.error(fail_message) return f"SYSTEM_ERROR: {fail_message}" @mcp.tool() @handle_anki_connection_error async def store_media_file( filename: str, url: str | None = None, data: str | None = None, path: str | None = None, ) -> str: """Store an image or media file in Anki's media folder. Use this tool to store images that can be referenced in flashcard fields using HTML img tags: <img src="filename"> This is useful when you need to: - Store an image before creating a note (e.g. to reference it in multiple notes) - Add an image to an existing card's field Provide exactly one of url, data, or path: - path: Absolute path to a local file. PREFERRED when the user shares an image file or screenshot. - url: A URL to download the image from (e.g. "https://example.com/photo.jpg") - data: Base64-encoded file content (for small images only) IMPORTANT: When a user shares an image file or screenshot, prefer using "path" with the absolute file path. AnkiConnect reads the file directly from disk, which avoids needing to base64-encode large image files. Args: filename: str - The filename to store the media as (e.g. "diagram.png"). url: str - Optional URL to download the image from. data: str - Optional base64-encoded image data. path: str - Optional absolute path to a local file. Preferred for user-shared files. """ if not url and not data and not path: return "SYSTEM_ERROR: Must provide either 'url', 'data', or 'path' for the media file." source = "path" if path else ("url" if url else "base64 data") async with get_anki_client() as anki: logger.info(f"Storing media file '{filename}' via {source}.") stored_filename = await anki.store_media_file( filename=filename, url=url, data=data, path=path, ) if stored_filename: return ( f"Successfully stored media file as '{stored_filename}'. " f'Reference it in card fields with: <img src="{stored_filename}">' ) else: return f"SYSTEM_ERROR: Failed to store media file '{filename}'." @mcp.tool() @handle_anki_connection_error async def search_notes( query: str = Field(description="Anki search query string"), limit: int = Field( default=20, ge=1, le=100, description="Maximum number of notes to return" ), ) -> str: """Search for notes in Anki using the powerful built-in search syntax. This tool allows you to find existing notes/flashcards using Anki's query language. Results include note IDs which can be used for follow-up actions. ## Common Search Patterns **Simple text search:** - `dog` - notes containing "dog" (matches "doggy", "underdog") - `dog cat` - notes with both "dog" AND "cat" - `dog or cat` - notes with "dog" OR "cat" - `-cat` - notes WITHOUT "cat" - `"a dog"` - exact phrase match - `w:dog` - whole word match only **Field-specific search:** - `front:dog` - Front field exactly equals "dog" - `front:*dog*` - Front field contains "dog" - `front:` - Front field is empty - `front:_*` - Front field is non-empty **Deck and tag filters:** - `deck:French` - cards in French deck (including subdecks) - `deck:French -deck:French::*` - only top-level French deck - `tag:vocab` - notes with "vocab" tag - `tag:none` - notes without any tags - `note:Basic` - notes using "Basic" note type **Card state:** - `is:due` - cards due for review - `is:new` - new cards not yet studied - `is:learn` - cards in learning phase - `is:review` - review cards - `is:suspended` - suspended cards - `is:buried` - buried cards **Card properties:** - `prop:ivl>=10` - interval >= 10 days - `prop:due=0` - due today - `prop:due=1` - due tomorrow - `prop:lapses>3` - lapsed more than 3 times - `prop:ease<2.5` - easier than default - `prop:reps<10` - reviewed fewer than 10 times **Recent activity:** - `added:7` - added in last 7 days - `edited:3` - edited in last 3 days - `rated:1` - answered today - `rated:7:1` - answered "Again" in last 7 days - `introduced:30` - first answered in last 30 days **Combining searches:** - `deck:Spanish tag:verb is:due` - due Spanish verbs - `added:7 -is:review` - new cards added this week - `(dog or cat) deck:Animals` - dog or cat in Animals deck Args: query: The Anki search query string. limit: Maximum notes to return (1-100, default 20). Returns: JSON array of matching notes with their fields, tags, and note IDs. """ async with get_anki_client() as anki: logger.debug(f"Searching notes with query: {query}") note_ids = await anki.find_notes(query=query) logger.info(f"Found {len(note_ids)} notes for query: {query}") if not note_ids: return json.dumps( { "query": query, "total_found": 0, "notes": [], "message": "No notes found matching the query.", }, indent=2, ) # Limit results limited_note_ids = note_ids[:limit] # Fetch note details notes_info = await anki.notes_info(limited_note_ids) # Format results formatted_results = _format_search_results(notes_info) result = { "query": query, "total_found": len(note_ids), "returned": len(formatted_results), "notes": formatted_results, } if len(note_ids) > limit: result["message"] = ( f"Showing {limit} of {len(note_ids)} matching notes. Refine your query or increase limit for more results." ) return json.dumps(result, indent=2, ensure_ascii=False)