principia-mcp

"""FastMCP server for Zotero library access."""

import sys

from fastmcp import FastMCP
from loguru import logger
from pydantic import BaseModel

from principia_mcp.models import Paper
from principia_mcp.search import SearchIndex
from principia_mcp.zotero import ZoteroReader

# Configure loguru - stderr so it doesn't interfere with MCP stdio
logger.remove()
logger.add(sys.stderr, level="INFO", format="{time:HH:mm:ss} | {level} | {message}")

mcp = FastMCP("principia-mcp")


# Lazy initialization of reader and index (allows env vars to be set first)
_reader: ZoteroReader | None = None
_index: SearchIndex | None = None


def get_reader() -> ZoteroReader:
    """Get or create the ZoteroReader instance."""
    global _reader
    if _reader is None:
        _reader = ZoteroReader()
    return _reader


def get_index() -> SearchIndex:
    """Get or create the SearchIndex, rebuilding if stale."""
    global _index
    if _index is None:
        _index = SearchIndex()

    reader = get_reader()
    mtime = reader.get_mtime()

    if _index.needs_rebuild(mtime) or _index.is_empty():
        logger.info("Building search index...")
        papers = reader.get_all_papers()
        _index.build(papers, mtime)

    return _index


# Response models for tool outputs
class CollectionInfo(BaseModel):
    """Collection information returned by tools."""

    id: int
    name: str
    item_count: int


class PaperInfo(BaseModel):
    """Paper information returned by tools."""

    id: int
    title: str
    authors: list[str]
    year: int | None
    abstract: str | None
    doi: str | None
    pdf_path: str | None
    collections: list[str]


def _paper_to_info(p: Paper) -> PaperInfo:
    """Convert Paper model to PaperInfo for API response."""
    abstract = p.abstract
    if abstract and len(abstract) > 500:
        abstract = abstract[:500] + "..."
    return PaperInfo(
        id=p.id,
        title=p.title,
        authors=p.authors,
        year=p.year,
        abstract=abstract,
        doi=p.doi,
        pdf_path=str(p.pdf_path) if p.pdf_path else None,
        collections=p.collections,
    )


@mcp.tool
def list_collections() -> list[CollectionInfo]:
    """List all Zotero collections with paper counts.

    Returns collections you can use to scope searches with the `collection` parameter.
    """
    reader = get_reader()
    collections = reader.list_collections()
    return [
        CollectionInfo(id=c.id, name=c.name, item_count=c.item_count) for c in collections
    ]


@mcp.tool
def list_papers(collection: str | None = None, limit: int = 50) -> list[PaperInfo]:
    """List papers in the library or a specific collection.

    Args:
        collection: Collection name to filter by (use list_collections to see available names)
        limit: Maximum papers to return (default 50)
    """
    reader = get_reader()
    papers = reader.list_papers(collection=collection, limit=limit)
    return [_paper_to_info(p) for p in papers]


@mcp.tool
def search_papers(
    query: str, collection: str | None = None, limit: int = 20
) -> list[PaperInfo]:
    """Search papers by keyword in title and abstract.

    Args:
        query: Search terms (searches in title and abstract)
        collection: Collection name to scope search (optional)
        limit: Maximum results (default 20)
    """
    reader = get_reader()
    index = get_index()

    # Get matching paper IDs from FTS5 index
    paper_ids = index.search(query=query, collection=collection, limit=limit)

    # Fetch full paper objects
    papers = []
    for paper_id in paper_ids:
        paper = reader.get_paper(paper_id)
        if paper:
            papers.append(paper)

    return [_paper_to_info(p) for p in papers]


@mcp.tool
def get_paper(paper_id: int) -> PaperInfo | None:
    """Get full paper details including PDF path.

    Use the pdf_path with the Read tool to read the paper content.
    Claude can read PDFs directly and see equations visually.

    Args:
        paper_id: Paper ID from search or list results
    """
    reader = get_reader()
    paper = reader.get_paper(paper_id)
    if paper is None:
        return None
    return _paper_to_info(paper)


def main():
    """Run the MCP server."""
    logger.info("Starting principia-mcp server")
    mcp.run()


if __name__ == "__main__":
    main()