Expense Tracker MCP Server

Instructions

Implementation Reference

• main.py:122-163 (handler)

  The MCP tool handler for 'get_item_history'. It calls the database helper query_item_history and formats the response, including handling empty results.

  @mcp.tool
  async def get_item_history(
      item_type: Annotated[str, "Item category to query (e.g., 'milk', 'bread', 'eggs')"],
      time_range_days: Annotated[
          int, "Number of days to look back (default: 365)"
      ] = 365,
  ) -> dict:
      """Query purchase history for a specific item type.
  
      Returns detailed purchase history including:
      - List of all purchases with dates, stores, quantities, and prices
      - Statistics: total purchases, date range, average frequency, total spent
  
      Args:
          item_type: Category to query (e.g., 'milk', 'bread', 'eggs')
          time_range_days: Number of days to look back (default: 365)
  
      Returns:
          Dictionary with purchases list and aggregated statistics
      """
      try:
          result = query_item_history(item_type, time_range_days)
  
          if not result["purchases"]:
              return {
                  "item_type": item_type,
                  "purchases": [],
                  "stats": {
                      "total_purchases": 0,
                      "message": f"No purchases found for '{item_type}' in the last {time_range_days} days",
                  },
              }
  
          return {
              "item_type": item_type,
              "purchases": result["purchases"],
              "stats": result["stats"],
          }
  
      except Exception as e:
          raise ToolError(f"Failed to query item history: {str(e)}")

• expense_tracker/database.py:137-223 (helper)

  Database helper function that executes SQL queries to retrieve purchase history and compute statistics for a given item type over a time range.

  def query_item_history(
      item_type: str,
      time_range_days: int = 365,
      db_path: Path = DEFAULT_DB_PATH,
  ) -> dict:
      """Query purchase history for a specific item type.
  
      Returns a dictionary with:
      - purchases: list of purchase records
      - stats: ItemStats object with aggregated statistics
      """
      cutoff_date = (datetime.now() - timedelta(days=time_range_days)).strftime(
          "%Y-%m-%d"
      )
  
      conn = get_connection(db_path)
  
      try:
          # Query purchases
          cursor = conn.execute(
              """
              SELECT
                  r.purchase_date,
                  r.store_name,
                  i.item_name_raw,
                  i.quantity,
                  i.unit_price,
                  i.line_total
              FROM items i
              JOIN receipts r ON i.receipt_id = r.id
              WHERE i.item_type = ? AND r.purchase_date >= ?
              ORDER BY r.purchase_date DESC
          """,
              (item_type, cutoff_date),
          )
  
          purchases = []
          for row in cursor.fetchall():
              purchases.append(
                  {
                      "date": row["purchase_date"],
                      "store": row["store_name"],
                      "item_name": row["item_name_raw"],
                      "quantity": row["quantity"],
                      "unit_price": row["unit_price"],
                      "price": row["line_total"],
                  }
              )
  
          # Query stats
          stats_cursor = conn.execute(
              """
              SELECT
                  COUNT(DISTINCT r.id) as total_purchases,
                  MAX(r.purchase_date) as last_purchase_date,
                  MIN(r.purchase_date) as first_purchase_date,
                  SUM(i.line_total) as total_spent
              FROM items i
              JOIN receipts r ON i.receipt_id = r.id
              WHERE i.item_type = ? AND r.purchase_date >= ?
          """,
              (item_type, cutoff_date),
          )
  
          stats_row = stats_cursor.fetchone()
  
          # Calculate average days between purchases
          avg_days = None
          if stats_row["total_purchases"] and stats_row["total_purchases"] > 1:
              first_date = datetime.fromisoformat(stats_row["first_purchase_date"])
              last_date = datetime.fromisoformat(stats_row["last_purchase_date"])
              total_days = (last_date - first_date).days
              if total_days > 0:
                  avg_days = total_days / (stats_row["total_purchases"] - 1)
  
          stats = {
              "total_purchases": stats_row["total_purchases"] or 0,
              "last_purchase_date": stats_row["last_purchase_date"],
              "first_purchase_date": stats_row["first_purchase_date"],
              "average_days_between": round(avg_days, 1) if avg_days else None,
              "total_spent": round(stats_row["total_spent"] or 0, 2),
          }
  
          return {"purchases": purchases, "stats": stats}
  
      finally:
          conn.close()