"""Example demonstrating the complete expense tracker workflow."""
import asyncio
from datetime import datetime
from expense_tracker.database import (
init_database,
insert_receipt,
insert_items,
query_item_history,
get_all_item_types,
)
from expense_tracker.models import Receipt, LineItem
from expense_tracker.categorizer import categorize_item
async def example_workflow():
"""Demonstrate the complete workflow without PDF parsing."""
print("=" * 70)
print("EXPENSE TRACKER MCP SERVER - DEMONSTRATION")
print("=" * 70)
# Step 1: Initialize database
print("\n1. Initializing database...")
init_database()
print(" ✓ Database initialized")
# Step 2: Create a sample receipt
print("\n2. Creating sample receipt...")
receipt = Receipt(
store_name="Walmart",
purchase_date="2025-01-15",
subtotal=42.50,
tax=2.55,
total=45.05,
)
receipt_id = insert_receipt(receipt)
print(f" ✓ Receipt #{receipt_id} created for {receipt.store_name}")
# Step 3: Add sample items (simulating parsed PDF items)
print("\n3. Categorizing and adding items...")
sample_items = [
{"name": "Organic Milk 2%", "price": 4.99},
{"name": "Whole Wheat Bread", "price": 3.49},
{"name": "Large Eggs Dozen", "price": 5.29},
{"name": "Fresh Broccoli", "price": 2.99},
{"name": "Gala Apples 3lb", "price": 4.99},
{"name": "Chicken Breast 2lb", "price": 12.99},
{"name": "Basmati Rice 5lb", "price": 8.99},
{"name": "Greek Yogurt", "price": 4.49},
]
categorized_items = []
for item in sample_items:
category = await categorize_item(item["name"], ctx=None)
line_item = LineItem(
item_name_raw=item["name"],
item_type=category,
line_total=item["price"],
quantity=1.0,
)
categorized_items.append(line_item)
print(f" - {item['name']:25} → {category}")
insert_items(receipt_id, categorized_items)
print(f"\n ✓ Added {len(categorized_items)} items to receipt")
# Step 4: Add another receipt (earlier date)
print("\n4. Adding second receipt (Costco from last month)...")
receipt2 = Receipt(
store_name="Costco",
purchase_date="2024-12-10",
subtotal=55.00,
tax=3.30,
total=58.30,
)
receipt_id2 = insert_receipt(receipt2)
costco_items = [
{"name": "Kirkland Milk 2gal", "price": 6.99},
{"name": "Organic Eggs 24ct", "price": 8.99},
{"name": "Basmati Rice 20lb", "price": 18.99},
]
categorized_items2 = []
for item in costco_items:
category = await categorize_item(item["name"], ctx=None)
line_item = LineItem(
item_name_raw=item["name"],
item_type=category,
line_total=item["price"],
quantity=1.0,
)
categorized_items2.append(line_item)
print(f" - {item['name']:25} → {category}")
insert_items(receipt_id2, categorized_items2)
print(f"\n ✓ Added {len(categorized_items2)} items to receipt")
# Step 5: Query item history
print("\n5. Querying purchase history...")
print("\n Query: 'When did I last buy milk?'")
milk_history = query_item_history("milk", time_range_days=365)
print(f" Answer: Last purchased on {milk_history['stats']['last_purchase_date']}")
print(f" Total purchases: {milk_history['stats']['total_purchases']}")
print(f" Total spent: ${milk_history['stats']['total_spent']:.2f}")
print("\n Purchase history:")
for purchase in milk_history["purchases"]:
print(
f" • {purchase['date']} at {purchase['store']} - "
f"{purchase['item_name']} (${purchase['price']:.2f})"
)
print("\n Query: 'Show me all rice purchases'")
rice_history = query_item_history("rice", time_range_days=365)
print(f" Total purchases: {rice_history['stats']['total_purchases']}")
print(f" Average days between purchases: {rice_history['stats']['average_days_between']:.1f}")
print("\n Purchase history:")
for purchase in rice_history["purchases"]:
print(
f" • {purchase['date']} at {purchase['store']} - "
f"{purchase['item_name']} (${purchase['price']:.2f})"
)
# Step 6: List all categories
print("\n6. Listing all tracked categories...")
all_types = get_all_item_types()
print(f"\n Found {len(all_types)} different item categories:\n")
for item_type in all_types:
print(
f" {item_type['item_type']:15} - "
f"{item_type['total_purchases']} purchases, "
f"last: {item_type['last_purchase_date']}, "
f"total spent: ${item_type['total_spent']:.2f}"
)
print("\n" + "=" * 70)
print("DEMONSTRATION COMPLETE")
print("=" * 70)
print("\nNext steps:")
print(" 1. Add this server to Claude Desktop configuration")
print(" 2. Upload a real PDF receipt")
print(" 3. Ask Claude questions about your expenses!")
print("=" * 70)
if __name__ == "__main__":
asyncio.run(example_workflow())