Expense Tracker MCP Server

Instructions

Implementation Reference

• main.py:27-120 (handler)

  Primary tool handler function. Decorated with @mcp.tool for registration. Implements PDF validation, parsing via pdf_parser, item categorization, database storage, and returns structured summary.

  @mcp.tool
  async def import_receipt_from_pdf(
      pdf_path: Annotated[str, "Absolute path to PDF receipt file"],
      ctx: Context,
  ) -> dict:
      """Import and parse a receipt from a PDF file.
  
      This tool:
      1. Extracts text from the PDF
      2. Parses receipt metadata (store, date, totals)
      3. Extracts line items with prices
      4. Categorizes each item using hybrid approach (static rules + LLM)
      5. Stores everything in SQLite database
  
      Args:
          pdf_path: Absolute path to the PDF receipt file
          ctx: FastMCP context for logging and LLM access
  
      Returns:
          Summary of imported receipt including store, date, item count, and category breakdown
      """
      try:
          await ctx.info(f"Starting import of receipt: {pdf_path}")
  
          # Validate path
          path = Path(pdf_path).expanduser().resolve()
          if not path.exists():
              raise ToolError(f"PDF file not found: {pdf_path}")
  
          if not path.suffix.lower() == ".pdf":
              raise ToolError(f"File must be a PDF: {pdf_path}")
  
          # Parse PDF
          await ctx.info("Extracting text from PDF...")
          receipt, raw_items = parse_pdf_receipt(path)
  
          await ctx.info(
              f"Parsed receipt: {receipt.store_name} on {receipt.purchase_date}"
          )
          await ctx.info(f"Found {len(raw_items)} line items")
  
          # Categorize items
          await ctx.info("Categorizing items...")
          categorized_items = []
          item_type_counts = {}
  
          for idx, item_dict in enumerate(raw_items):
              # Categorize using hybrid approach
              item_type = await categorize_item(item_dict["item_name"], ctx)
  
              # Create LineItem object
              line_item = LineItem(
                  item_name_raw=item_dict["item_name"],
                  item_type=item_type,
                  quantity=item_dict["quantity"],
                  line_total=item_dict["price"],
              )
  
              categorized_items.append(line_item)
  
              # Track category counts
              item_type_counts[item_type] = item_type_counts.get(item_type, 0) + 1
  
              await ctx.debug(
                  f"  [{idx+1}/{len(raw_items)}] {item_dict['item_name']} -> {item_type}"
              )
  
          # Insert into database
          await ctx.info("Saving to database...")
          receipt_id = insert_receipt(receipt)
          insert_items(receipt_id, categorized_items)
  
          await ctx.info(f"Successfully imported receipt #{receipt_id}")
  
          # Return summary
          return {
              "status": "success",
              "receipt_id": receipt_id,
              "store_name": receipt.store_name,
              "purchase_date": receipt.purchase_date,
              "total": receipt.total,
              "items_count": len(categorized_items),
              "item_types": item_type_counts,
              "message": f"Successfully imported {len(categorized_items)} items from {receipt.store_name}",
          }
  
      except FileNotFoundError as e:
          raise ToolError(f"File not found: {str(e)}")
      except ValueError as e:
          raise ToolError(f"Failed to parse receipt: {str(e)}")
      except Exception as e:
          await ctx.error(f"Unexpected error during import: {e}")
          raise ToolError(f"Failed to import receipt: {str(e)}")

• expense_tracker/models.py:7-54 (schema)

  Dataclass models for Receipt and LineItem providing structured data validation and typing for parsed receipt data used throughout the tool.

  @dataclass
  class Receipt:
      """Represents a parsed receipt."""
  
      store_name: str
      purchase_date: str  # ISO format: YYYY-MM-DD
      total: float
      subtotal: Optional[float] = None
      tax: Optional[float] = None
  
      def __post_init__(self):
          """Validate receipt data."""
          if self.total <= 0:
              raise ValueError("Total must be positive")
  
  
  @dataclass
  class LineItem:
      """Represents a single item from a receipt."""
  
      item_name_raw: str
      item_type: str
      line_total: float
      quantity: float = 1.0
      unit_price: Optional[float] = None
  
      def __post_init__(self):
          """Calculate unit price if not provided."""
          if self.unit_price is None and self.quantity > 0:
              self.unit_price = self.line_total / self.quantity
  
          if self.line_total <= 0:
              raise ValueError("Line total must be positive")
          if self.quantity <= 0:
              raise ValueError("Quantity must be positive")
  
  
  @dataclass
  class ItemStats:
      """Statistics for a specific item type."""
  
      item_type: str
      total_purchases: int
      last_purchase_date: str
      first_purchase_date: str
      total_spent: float
      average_days_between: Optional[float] = None

• expense_tracker/pdf_parser.py:268-290 (helper)

  Key helper function that extracts text from PDF using pdfplumber, parses store/date/totals/line items using regex patterns, returns Receipt and raw items list.

  def parse_pdf_receipt(pdf_path: Path) -> tuple[Receipt, list[dict]]:
      """Parse a PDF receipt file.
  
      Args:
          pdf_path: Path to PDF file
  
      Returns:
          Tuple of (Receipt object, list of item dicts)
  
      Raises:
          FileNotFoundError: If PDF doesn't exist
          ValueError: If parsing fails
      """
      # Convert string to Path if needed
      if isinstance(pdf_path, str):
          pdf_path = Path(pdf_path)
  
      # Extract text
      text = extract_text_from_pdf(pdf_path)
  
      # Parse receipt
      return parse_receipt(text)

• expense_tracker/categorizer.py:221-243 (helper)

  Hybrid item categorization helper: static regex/pattern matching first, LLM fallback via ctx.sample for unknown items.

  async def categorize_item(item_name: str, ctx=None) -> str:
      """Main categorization function with hybrid approach.
  
      Args:
          item_name: Raw item name from receipt
          ctx: Optional FastMCP Context for LLM fallback
  
      Returns:
          item_type category (guaranteed to return a value)
      """
      # Try deterministic rules first
      category = deterministic_categorize(item_name)
  
      if category:
          return category
  
      # Fall back to LLM if context is available
      if ctx:
          return await llm_categorize(item_name, ctx)
  
      # Ultimate fallback
      return "other"

• expense_tracker/database.py:81-135 (helper)

  Database persistence helpers: insert_receipt creates receipt record, insert_items adds categorized line items with foreign key.

  def insert_receipt(receipt: Receipt, db_path: Path = DEFAULT_DB_PATH) -> int:
      """Insert a receipt and return its ID."""
      conn = get_connection(db_path)
  
      try:
          cursor = conn.execute(
              """
              INSERT INTO receipts (store_name, purchase_date, subtotal, tax, total)
              VALUES (?, ?, ?, ?, ?)
          """,
              (
                  receipt.store_name,
                  receipt.purchase_date,
                  receipt.subtotal,
                  receipt.tax,
                  receipt.total,
              ),
          )
          conn.commit()
          return cursor.lastrowid
      finally:
          conn.close()
  
  
  def insert_items(
      receipt_id: int, items: list[LineItem], db_path: Path = DEFAULT_DB_PATH
  ) -> None:
      """Bulk insert items for a receipt."""
      if not items:
          return
  
      conn = get_connection(db_path)
  
      try:
          conn.executemany(
              """
              INSERT INTO items (receipt_id, item_name_raw, item_type, quantity, unit_price, line_total)
              VALUES (?, ?, ?, ?, ?, ?)
          """,
              [
                  (
                      receipt_id,
                      item.item_name_raw,
                      item.item_type,
                      item.quantity,
                      item.unit_price,
                      item.line_total,
                  )
                  for item in items
              ],
          )
          conn.commit()
      finally:
          conn.close()