"""MCP server for Anki flashcard operations."""
import asyncio
import json
from typing import Any
from mcp.server import Server
from mcp.types import Tool, TextContent, ErrorData
import mcp.server.stdio
from .anki_client import AnkiConnectClient
# Initialize server and client
server = Server("mcp-server-anki")
anki_client = AnkiConnectClient()
@server.list_tools()
async def list_tools() -> list[Tool]:
"""List available MCP tools."""
return [
Tool(
name="list_decks",
description="List all Anki decks",
inputSchema={
"type": "object",
"properties": {},
"required": []
}
),
Tool(
name="list_cards",
description="List all flashcards in a specific deck",
inputSchema={
"type": "object",
"properties": {
"deck_name": {
"type": "string",
"description": "Name of the deck to list cards from"
}
},
"required": ["deck_name"]
}
),
Tool(
name="create_deck",
description="Create a new Anki deck",
inputSchema={
"type": "object",
"properties": {
"deck_name": {
"type": "string",
"description": "Name of the deck to create"
}
},
"required": ["deck_name"]
}
),
Tool(
name="create_card",
description="Create a new flashcard (note) in a deck",
inputSchema={
"type": "object",
"properties": {
"deck_name": {
"type": "string",
"description": "Name of the deck to add card to"
},
"front": {
"type": "string",
"description": "Front side of the card (question/prompt)"
},
"back": {
"type": "string",
"description": "Back side of the card (answer)"
},
"tags": {
"type": "array",
"items": {"type": "string"},
"description": "Optional tags for the card"
}
},
"required": ["deck_name", "front", "back"]
}
)
]
@server.call_tool()
async def call_tool(name: str, arguments: Any) -> list[TextContent]:
"""Handle tool execution requests."""
try:
if name == "list_decks":
return await handle_list_decks()
elif name == "list_cards":
return await handle_list_cards(arguments)
elif name == "create_deck":
return await handle_create_deck(arguments)
elif name == "create_card":
return await handle_create_card(arguments)
else:
raise ValueError(f"Unknown tool: {name}")
except ConnectionError as e:
return [TextContent(
type="text",
text=f"Error: {str(e)}"
)]
except ValueError as e:
return [TextContent(
type="text",
text=f"Error: {str(e)}"
)]
except RuntimeError as e:
return [TextContent(
type="text",
text=f"Error: {str(e)}"
)]
except Exception as e:
return [TextContent(
type="text",
text=f"Unexpected error: {str(e)}"
)]
async def handle_list_decks() -> list[TextContent]:
"""Handle list_decks tool."""
decks = anki_client.get_deck_names()
result = {"decks": decks}
return [TextContent(
type="text",
text=json.dumps(result, indent=2)
)]
async def handle_list_cards(arguments: dict) -> list[TextContent]:
"""Handle list_cards tool."""
deck_name = arguments.get("deck_name")
if not deck_name:
raise ValueError("deck_name is required")
cards = anki_client.get_cards_in_deck(deck_name)
result = {"cards": cards}
return [TextContent(
type="text",
text=json.dumps(result, indent=2)
)]
async def handle_create_deck(arguments: dict) -> list[TextContent]:
"""Handle create_deck tool."""
deck_name = arguments.get("deck_name")
if not deck_name:
raise ValueError("deck_name is required")
deck_id = anki_client.create_deck(deck_name)
result = {
"deck_id": deck_id,
"message": f"Deck '{deck_name}' created successfully"
}
return [TextContent(
type="text",
text=json.dumps(result, indent=2)
)]
async def handle_create_card(arguments: dict) -> list[TextContent]:
"""Handle create_card tool."""
deck_name = arguments.get("deck_name")
front = arguments.get("front")
back = arguments.get("back")
tags = arguments.get("tags", [])
if not deck_name:
raise ValueError("deck_name is required")
if not front:
raise ValueError("front is required")
if not back:
raise ValueError("back is required")
note_id = anki_client.create_card(deck_name, front, back, tags)
result = {
"note_id": note_id,
"message": f"Card created successfully in deck '{deck_name}'"
}
return [TextContent(
type="text",
text=json.dumps(result, indent=2)
)]
async def main():
"""Run the MCP server."""
async with mcp.server.stdio.stdio_server() as (read_stream, write_stream):
await server.run(
read_stream,
write_stream,
server.create_initialization_options()
)
def run():
"""Entry point for the server."""
asyncio.run(main())
if __name__ == "__main__":
run()
