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()