MCP-AnkiConnect
by samefarrar
- 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()