MCP Agent - AI Expense Tracker

#!/usr/bin/env python3 """Expense Tracker API - Main Application""" from fastapi import FastAPI, HTTPException, Query from fastapi.middleware.cors import CORSMiddleware from fastapi.responses import HTMLResponse from typing import List, Optional from datetime import datetime from pathlib import Path import uvicorn from models import ( Transaction, TransactionCreate, TransactionUpdate, TransactionType, Category, TransactionSummary, CategorySummary ) from store import TransactionStore from config import settings # Create FastAPI application app = FastAPI( title=settings.API_TITLE, version=settings.API_VERSION, docs_url="/docs", redoc_url="/redoc" ) # Add CORS middleware app.add_middleware( CORSMiddleware, allow_origins=settings.ALLOW_ORIGINS, allow_credentials=settings.ALLOW_CREDENTIALS, allow_methods=settings.ALLOW_METHODS, allow_headers=settings.ALLOW_HEADERS, ) # Initialize transaction store transaction_store = TransactionStore() @app.get("/", response_class=HTMLResponse) async def index(): """Serve the dashboard HTML page""" template_path = Path(__file__).parent / "templates" / "index.html" with open(template_path, "r", encoding="utf-8") as f: return f.read() @app.get("/health") async def health_check(): """Health check endpoint""" return { "status": "healthy", "service": "expense-tracker-api", "version": settings.API_VERSION, "timestamp": datetime.now().isoformat(), "transactions_count": len(transaction_store.get_all()) } @app.get("/transactions", response_model=List[Transaction]) async def get_transactions( skip: int = Query(0, ge=0), limit: int = Query(50, ge=1, le=500), type: Optional[TransactionType] = None, category: Optional[Category] = None ): """Get all transactions with pagination and filtering""" transactions = transaction_store.get_all() # Apply filters if type: transactions = [t for t in transactions if t.type == type] if category: transactions = [t for t in transactions if t.category == category] return transactions[skip:skip + limit] @app.get("/transactions/search", response_model=List[Transaction]) async def search_transactions(q: str = Query(..., min_length=1)): """Search transactions by title, description, category, or tags""" return transaction_store.search(q) @app.get("/transactions/{transaction_id}", response_model=Transaction) async def get_transaction(transaction_id: str): """Get a specific transaction by ID""" transaction = transaction_store.get_by_id(transaction_id) if not transaction: raise HTTPException(status_code=404, detail="Transaction not found") return transaction @app.post("/transactions", response_model=Transaction, status_code=201) async def create_transaction(transaction_data: TransactionCreate): """Create a new transaction""" return transaction_store.create(transaction_data) @app.put("/transactions/{transaction_id}", response_model=Transaction) async def update_transaction(transaction_id: str, transaction_data: TransactionUpdate): """Update an existing transaction""" if not transaction_store.get_by_id(transaction_id): raise HTTPException(status_code=404, detail="Transaction not found") return transaction_store.update(transaction_id, transaction_data) @app.delete("/transactions/{transaction_id}") async def delete_transaction(transaction_id: str): """Delete a transaction""" if not transaction_store.delete(transaction_id): raise HTTPException(status_code=404, detail="Transaction not found") return {"message": "Transaction deleted successfully", "transaction_id": transaction_id} @app.get("/summary", response_model=TransactionSummary) async def get_summary(): """Get transaction summary statistics""" summary_data = transaction_store.get_summary() return TransactionSummary(**summary_data) @app.get("/summary/categories", response_model=List[CategorySummary]) async def get_category_summary(): """Get summary by category""" category_data = transaction_store.get_category_summary() return [CategorySummary(**item) for item in category_data] @app.get("/transactions/type/{transaction_type}", response_model=List[Transaction]) async def get_by_type(transaction_type: TransactionType): """Get transactions by type (income or expense)""" return transaction_store.get_by_type(transaction_type) @app.get("/transactions/category/{category}", response_model=List[Transaction]) async def get_by_category(category: Category): """Get transactions by category""" return transaction_store.get_by_category(category) @app.get("/transactions/date-range") async def get_by_date_range( start_date: datetime = Query(...), end_date: datetime = Query(...) ): """Get transactions within a date range""" if start_date > end_date: raise HTTPException(status_code=400, detail="Start date must be before end date") return transaction_store.get_by_date_range(start_date, end_date) if __name__ == "__main__": print(f"🚀 Starting Expense Tracker API...") print(f"🌐 Server: http://{settings.HOST}:{settings.PORT}") print(f"📚 API Docs: http://{settings.HOST}:{settings.PORT}/docs") uvicorn.run( "main:app", host=settings.HOST, port=settings.PORT, reload=True )