MCP-AnkiConnect

  • mcp_ankiconnect
from typing import Any, List, Optional import logging import httpx import asyncio from enum import Enum try: from mcp_ankiconnect.config import ( ANKI_CONNECT_URL, ANKI_CONNECT_VERSION, TIMEOUTS, ) except ImportError: from config import ( ANKI_CONNECT_URL, ANKI_CONNECT_VERSION, TIMEOUTS, ) from pydantic import BaseModel, Field logger = logging.getLogger(__name__) class AnkiAction(str, Enum): DECK_NAMES = "deckNames" FIND_CARDS = "findCards" CARDS_INFO = "cardsInfo" ANSWER_CARDS = "answerCards" MODEL_NAMES = "modelNames" MODEL_FIELD_NAMES = "modelFieldNames" ADD_NOTE = "addNote" FIND_NOTES = "findNotes" NOTES_INFO = "notesInfo" class AnkiConnectResponse(BaseModel): result: Any error: Optional[str] = None class AnkiConnectRequest(BaseModel): action: AnkiAction version: int = 6 params: dict = Field(default_factory=dict) class AnkiConnectClient: def __init__(self, base_url: str = ANKI_CONNECT_URL): self.base_url = base_url self.client = httpx.AsyncClient(timeout=TIMEOUTS) logger.info(f"Initialized AnkiConnect client with base URL: {base_url}") async def invoke(self, action: str, **params) -> Any: request = AnkiConnectRequest( action=action, version=ANKI_CONNECT_VERSION, params=params ) logger.debug(f"Invoking AnkiConnect action: {action} with params: {params}") retries = 3 for attempt in range(retries): try: response = await self.client.post( self.base_url, json=request.model_dump() ) break except httpx.TimeoutException as e: if attempt == retries - 1: raise RuntimeError(f"Unable to connect to Anki after {retries} attempts. Please ensure Anki is running and the AnkiConnect plugin is installed.") await asyncio.sleep(2 ** attempt) # Exponential backoff: 1, 2, 4 seconds continue try: response.raise_for_status() anki_response = AnkiConnectResponse.model_validate(response.json()) if anki_response.error: logger.error(f"AnkiConnect error for action {action}: {anki_response.error}") raise ValueError(f"AnkiConnect error: {anki_response.error}") return anki_response.result except httpx.HTTPError as e: logger.error(f"HTTP error while invoking {action}: {str(e)}") raise RuntimeError(f"Failed to communicate with AnkiConnect: {str(e)}") from e except ValueError as e: # Re-raise ValueError (from AnkiConnect errors) directly raise except Exception as e: logger.error(f"Unexpected error while invoking {action}: {str(e)}") raise RuntimeError(f"Unexpected error during AnkiConnect operation: {str(e)}") from e async def cards_info(self, card_ids: List[int]) -> List[dict]: try: return await self.invoke(AnkiAction.CARDS_INFO, cards=card_ids) except Exception as e: raise RuntimeError(f"Error getting cards info: {str(e)}") from e async def deck_names(self) -> List[str]: try: return await self.invoke(AnkiAction.DECK_NAMES) except Exception as e: if isinstance(e, RuntimeError) and "Unable to connect to Anki" in str(e): raise raise RuntimeError(f"Error getting deck names: {str(e)}") from e async def find_cards(self, query: str) -> List[int]: try: return await self.invoke(AnkiAction.FIND_CARDS, query=query) except Exception as e: raise RuntimeError(f"Error finding cards: {str(e)}") from e async def answer_cards(self, answers: List[dict]) -> List[bool]: try: return await self.invoke(AnkiAction.ANSWER_CARDS, answers=answers) except Exception as e: raise RuntimeError(f"Error answering cards: {str(e)}") from e async def model_field_names(self, model_name: str) -> List[str]: try: return await self.invoke(AnkiAction.MODEL_FIELD_NAMES, modelName=model_name) except Exception as e: raise RuntimeError(f"Error getting model field names: {str(e)}") from e async def model_names(self) -> List[str]: try: return await self.invoke(AnkiAction.MODEL_NAMES) except Exception as e: raise RuntimeError(f"Error getting model names: {str(e)}") from e async def find_notes(self, query: str) -> List[int]: try: return await self.invoke(AnkiAction.FIND_NOTES, query=query) except Exception as e: raise RuntimeError(f"Error finding notes: {str(e)}") from e async def add_note(self, note: dict) -> int: try: return await self.invoke(AnkiAction.ADD_NOTE, note=note) except Exception as e: raise RuntimeError(f"Error adding note: {str(e)}") from e async def notes_info(self, note_ids: List[int]) -> List[dict]: try: return await self.invoke(AnkiAction.NOTES_INFO, notes=note_ids) except Exception as e: raise RuntimeError(f"Error getting notes info: {str(e)}") from e async def close(self): await self.client.aclose()