Neolibrarian MCP

#!/usr/bin/env python3 # -*- coding: utf-8 -*- """Calibre Library MCP Server Modern MCP server exposing Calibre library tools and a few read-only resources. Uses the FastMCP interface from the official MCP Python SDK with a lifespan initializer for clean dependency management. """ import logging import os from dataclasses import dataclass from typing import Any, Optional, Dict, List from collections.abc import AsyncIterator from contextlib import asynccontextmanager from mcp.server.fastmcp import FastMCP, Context from calibre_client import CalibreClient # Configure structured logging logging.basicConfig( level=logging.INFO, format="%(asctime)s - %(name)s - %(levelname)s - %(message)s", handlers=[ logging.StreamHandler(), ] ) logger = logging.getLogger("calibre-mcp-server") @dataclass class AppContext: """Typed lifespan context for request handlers.""" client: CalibreClient @asynccontextmanager async def app_lifespan(_server: FastMCP) -> AsyncIterator[AppContext]: """Initialize Calibre client once and clean up on shutdown.""" logger.info("Starting Calibre MCP server...") # Determine config path relative to this file try: script_dir = os.path.dirname(os.path.abspath(__file__)) logger.debug(f"Script directory: {script_dir}") except NameError: script_dir = os.getcwd() logger.debug(f"Using current working directory: {script_dir}") # Allow override via environment variable for easy external configuration env_cfg = os.environ.get("CALIBRE_CONFIG_PATH") config_path = ( os.path.expanduser(os.path.expandvars(env_cfg)) if env_cfg and env_cfg.strip() else os.path.join(script_dir, "config.json") ) logger.info(f"Loading configuration from: {config_path}") # Initialize client client: Optional[CalibreClient] = None try: client = CalibreClient(config_path) logger.info("Calibre client initialized successfully") yield AppContext(client=client) except Exception as e: logger.error(f"Failed to initialize Calibre client: {e}") raise finally: # No explicit cleanup required for CalibreClient currently logger.info("Calibre MCP server shutting down") # Create FastMCP server with lifespan lifecycle mcp = FastMCP("Calibre Library Server", lifespan=app_lifespan) # Helpful alias for typed context; generic parameters are managed by the SDK CalCtx = Context def get_client(ctx: CalCtx) -> CalibreClient: """Helper function to get client from context.""" return ctx.request_context.lifespan_context.client @mcp.tool() def ping(ctx: CalCtx) -> Dict[str, Any]: """Test connection to Calibre library.""" return get_client(ctx).ping() @mcp.tool() def get_total_books(ctx: CalCtx) -> Dict[str, Any]: """Get total number of books in the library.""" return get_client(ctx).get_total_books() @mcp.tool() def search_by_author(author_name: str, ctx: CalCtx) -> Dict[str, Any]: """Search for books by author name (case-insensitive, partial match). Args: author_name: Author name to search for """ if not author_name.strip(): return {"status": "error", "message": "Author name cannot be empty"} return get_client(ctx).search_by_author(author_name.strip()) @mcp.tool() def search_by_title(title: str, ctx: CalCtx) -> Dict[str, Any]: """Search for books by title (case-insensitive, partial match). Args: title: Book title to search for """ if not title.strip(): return {"status": "error", "message": "Title cannot be empty"} return get_client(ctx).search_by_title(title.strip()) @mcp.tool() def get_book_details(book_id: int, ctx: CalCtx) -> Dict[str, Any]: """Get detailed metadata for a specific book by book_id. Args: book_id: The book ID to get details for """ if book_id <= 0: return {"status": "error", "message": "Book ID must be positive"} return get_client(ctx).get_book_details(book_id) @mcp.tool() def get_book_formats(book_id: int, ctx: CalCtx) -> Dict[str, Any]: """Get all available formats (EPUB, PDF, TXT, MOBI) for a book. Args: book_id: The book ID to get formats for """ if book_id <= 0: return {"status": "error", "message": "Book ID must be positive"} return get_client(ctx).get_book_formats(book_id) @mcp.tool() def get_book_content( book_id: int, ctx: CalCtx, preferred_format: Optional[str] = None, max_length: int = 50000, ) -> Dict[str, Any]: """Extract full text content from a book file. Args: book_id: The book ID to extract content from preferred_format: Preferred format (EPUB, PDF, TXT, MOBI) max_length: Maximum content length in characters (default: 50000) """ if book_id <= 0: return {"status": "error", "message": "Book ID must be positive"} if max_length <= 0 or max_length > 500000: return {"status": "error", "message": "max_length must be between 1 and 500000"} if preferred_format and preferred_format.upper() not in ["EPUB", "PDF", "TXT", "MOBI"]: return {"status": "error", "message": "preferred_format must be one of: EPUB, PDF, TXT, MOBI"} return get_client(ctx).get_book_content(book_id, preferred_format, max_length) @mcp.tool() def get_book_sample( book_id: int, ctx: CalCtx, sample_type: str = "beginning", sample_size: int = 5000, ) -> Dict[str, Any]: """Get a sample of book content for analysis. Args: book_id: The book ID to sample from sample_type: Type of sample to extract (beginning, end, middle, overview) sample_size: Sample size in characters (default: 5000) """ if book_id <= 0: return {"status": "error", "message": "Book ID must be positive"} if sample_size <= 0 or sample_size > 50000: return {"status": "error", "message": "sample_size must be between 1 and 50000"} valid_types = ["beginning", "end", "middle", "overview"] if sample_type not in valid_types: return {"status": "error", "message": f"sample_type must be one of: {', '.join(valid_types)}"} return get_client(ctx).get_book_sample(book_id, sample_type, sample_size) @mcp.tool() def analyze_book_content( book_id: int, ctx: CalCtx, analysis_type: str = "summary", max_length: int = 5000, ) -> Dict[str, Any]: """Get book content prepared for specific types of LLM analysis. Args: book_id: The book ID to analyze analysis_type: Type of analysis to prepare content for (summary, beginning, themes, characters, quotes) max_length: Maximum content length for analysis (default: 5000) """ if book_id <= 0: return {"status": "error", "message": "Book ID must be positive"} if max_length <= 0 or max_length > 50000: return {"status": "error", "message": "max_length must be between 1 and 50000"} valid_types = ["summary", "beginning", "themes", "characters", "quotes"] if analysis_type not in valid_types: return {"status": "error", "message": f"analysis_type must be one of: {', '.join(valid_types)}"} return get_client(ctx).analyze_book_content(book_id, analysis_type, max_length) @mcp.tool() def search_content( book_id: int, query: str, ctx: CalCtx, case_sensitive: bool = False, ) -> Dict[str, Any]: """Search for text within a book's content. Args: book_id: The book ID to search in query: Text to search for case_sensitive: Whether the search should be case sensitive (default: False) """ if book_id <= 0: return {"status": "error", "message": "Book ID must be positive"} if not query.strip(): return {"status": "error", "message": "Search query cannot be empty"} return get_client(ctx).search_content(book_id, query.strip(), case_sensitive) @mcp.tool() def search_multiple_books( book_ids: List[int], query: str, ctx: CalCtx, case_sensitive: bool = False, ) -> Dict[str, Any]: """Search for text within multiple books' content. Args: book_ids: List of book IDs to search in query: Text to search for case_sensitive: Whether the search should be case sensitive (default: False) """ if not book_ids or not isinstance(book_ids, list): return {"status": "error", "message": "book_ids must be a non-empty list"} if len(book_ids) > 100: return {"status": "error", "message": "Too many books (max 100)"} invalid_ids = [bid for bid in book_ids if not isinstance(bid, int) or bid <= 0] if invalid_ids: return {"status": "error", "message": f"Invalid book IDs: {invalid_ids}"} if not query.strip(): return {"status": "error", "message": "Search query cannot be empty"} return get_client(ctx).search_multiple_books(book_ids, query.strip(), case_sensitive) @mcp.tool() def unified_search( ctx: CalCtx, query: Optional[str] = None, author: Optional[str] = None, title: Optional[str] = None, series: Optional[str] = None, formats: Optional[List[str]] = None, date_start: Optional[str] = None, date_end: Optional[str] = None, offset: int = 0, limit: int = 20, sort_by: str = "title", fuzzy: bool = False ) -> Dict[str, Any]: """Unified search with comprehensive filters and pagination. Args: query: General search term (searches both title and author) author: Filter by author name title: Filter by title series: Filter by series name formats: List of formats to filter by (e.g., ["EPUB", "PDF"]) date_start: Start date for publication date range (YYYY-MM-DD) date_end: End date for publication date range (YYYY-MM-DD) offset: Number of results to skip (default: 0) limit: Maximum results to return (default: 20, max: 100) sort_by: Sort order: title, author, date, or series (default: title) fuzzy: Enable fuzzy matching for typos (default: False) """ if limit > 100: limit = 100 filters = {} if author: filters["author"] = author.strip() if title: filters["title"] = title.strip() if series: filters["series"] = series.strip() if formats: filters["formats"] = formats if date_start or date_end: date_range = {} if date_start: date_range["start"] = date_start if date_end: date_range["end"] = date_end filters["date_range"] = date_range pagination = {"offset": offset, "limit": limit} options = {"sort_by": sort_by, "fuzzy_matching": fuzzy} return get_client(ctx).unified_search( query=query.strip() if query else "", filters=filters, pagination=pagination, options=options ) @mcp.tool() def get_books_batch(book_ids: List[int], ctx: CalCtx) -> Dict[str, Any]: """Get detailed metadata for multiple books in a single request. Args: book_ids: List of book IDs to retrieve (max 100) """ if not book_ids or not isinstance(book_ids, list): return {"status": "error", "message": "book_ids must be a non-empty list"} if len(book_ids) > 100: return {"status": "error", "message": "Maximum 100 books per batch request"} invalid_ids = [bid for bid in book_ids if not isinstance(bid, int) or bid <= 0] if invalid_ids: return {"status": "error", "message": f"Invalid book IDs: {invalid_ids}"} return get_client(ctx).get_books_batch(book_ids) @mcp.tool() def get_random_books( ctx: CalCtx, count: int = 10, formats: Optional[List[str]] = None, series_only: bool = False ) -> Dict[str, Any]: """Get random books for discovery. Args: count: Number of random books to return (default: 10, max: 50) formats: Only include books with these formats (e.g., ["EPUB", "PDF"]) series_only: Only include books that are part of a series (default: False) """ if count <= 0 or count > 50: count = min(max(count, 1), 50) filters = {} if formats: filters["formats"] = formats if series_only: filters["series_only"] = True return get_client(ctx).get_random_books(count, filters) @mcp.tool() def full_text_search( query: str, ctx: CalCtx, case_sensitive: bool = False, max_results: int = 50, context_chars: int = 200 ) -> Dict[str, Any]: """Search for text content across all books using full-text search database. Args: query: Text to search for across all book content case_sensitive: Whether the search should be case sensitive (default: False) max_results: Maximum number of results to return (default: 50, max: 200) context_chars: Number of characters to include around matches for context (default: 200) """ if not query.strip(): return {"status": "error", "message": "Search query cannot be empty"} if max_results <= 0 or max_results > 200: max_results = min(max(max_results, 1), 200) if context_chars < 50 or context_chars > 1000: context_chars = min(max(context_chars, 50), 1000) try: result = get_client(ctx).full_text_search(query.strip(), case_sensitive, max_results) # Add search parameters to response for transparency if result.get("status") == "success": result["search_parameters"] = { "query": query.strip(), "case_sensitive": case_sensitive, "max_results": max_results, "context_chars": context_chars, "results_truncated": len(result.get("results", [])) >= max_results } return result except Exception as e: return { "status": "error", "message": f"Full-text search failed: {str(e)}", "error_type": "search_error" } @mcp.tool() def full_text_search_book( book_id: int, query: str, ctx: CalCtx, case_sensitive: bool = False, context_chars: int = 200 ) -> Dict[str, Any]: """Search for text content within a specific book using full-text search database. Args: book_id: The book ID to search within query: Text to search for case_sensitive: Whether the search should be case sensitive (default: False) context_chars: Number of characters to include around matches for context (default: 200) """ if book_id <= 0: return {"status": "error", "message": "Book ID must be positive"} if not query.strip(): return {"status": "error", "message": "Search query cannot be empty"} if context_chars < 50 or context_chars > 1000: context_chars = min(max(context_chars, 50), 1000) try: result = get_client(ctx).full_text_search_book(book_id, query.strip(), case_sensitive) # Add search parameters to response for transparency if result.get("status") == "success": result["search_parameters"] = { "book_id": book_id, "query": query.strip(), "case_sensitive": case_sensitive, "context_chars": context_chars } return result except Exception as e: return { "status": "error", "message": f"Book full-text search failed: {str(e)}", "error_type": "search_error" } @mcp.tool() def get_full_text_search_stats(ctx: CalCtx) -> Dict[str, Any]: """Get statistics about the full-text search database. Returns information about indexed books, formats, and search capabilities. """ try: result = get_client(ctx).get_full_text_search_stats() if result.get("status") == "success": result["capabilities"] = { "global_search": True, "book_specific_search": True, "case_sensitive_search": True, "context_extraction": True } return result except Exception as e: return { "status": "error", "message": f"Failed to get FTS statistics: {str(e)}", "error_type": "stats_error" } # --- Resources: lightweight, read-only context surfaces --- @mcp.resource("calibre://stats", mime_type="application/json") def resource_stats() -> Dict[str, Any]: """Library stats overview as JSON.""" ctx = mcp.get_context() client = get_client(ctx) ping = client.ping() total = client.get_total_books() return { "status": ping.get("status", "unknown"), "library_path": ping.get("library_path", "unknown"), "books": total.get("total", total), "fts_db": ping.get("checks", {}).get("fts_db", "unknown"), } @mcp.resource("calibre://book/{book_id}/details", mime_type="application/json") def resource_book_details(book_id: int) -> Dict[str, Any]: """JSON details for a specific book.""" if book_id <= 0: return {"status": "error", "message": "Book ID must be positive"} ctx = mcp.get_context() return get_client(ctx).get_book_details(book_id) if __name__ == "__main__": # Explicitly use stdio transport for MCP hosts like Claude Desktop mcp.run(transport="stdio")