add_expense

Instructions

Implementation Reference

• main.py:45-60 (handler)

  The main handler function for the 'add_expense' tool. Decorated with @mcp.tool, it accepts description, amount, category, subcategory (optional), and note (optional). Inserts the expense into a SQLite database and returns a dict with status and the new record's id.

  async def add_expense(description: str, amount: float, category: str, subcategory: str = '', note: str = '') -> dict:
      """ Add an expense to the database """
      try:
          async with aiosqlite.connect(DB_PATH) as conn:
              now = datetime.now().isoformat()
              cursor = await conn.execute("""
              INSERT INTO expenses (description, amount, category, subcategory, date, note, created_at, updated_at)
              VALUES (?, ?, ?, ?, ?, ?, ?, ?)
              """, (description, amount, category, subcategory, datetime.now().strftime('%Y-%m-%d'), note, now, now))
              last_id = cursor.lastrowid
              await conn.commit()
              return {"status": "OK", "id": last_id}
      except aiosqlite.Error as e:
          return {"status": "ERROR", "message": f"Database error: {str(e)}"}
      except Exception as e:
          return {"status": "ERROR", "message": f"Unexpected error: {str(e)}"}

• main.py:44-44 (registration)

  Tool registration via the @mcp.tool decorator. The FastMCP instance 'mcp' registers 'add_expense' as an MCP tool.

  @mcp.tool

• main.py:45-45 (schema)

  Input parameter definitions (type hints): description (str), amount (float), category (str), subcategory (str, default ''), note (str, default ''). Output is dict.

  async def add_expense(description: str, amount: float, category: str, subcategory: str = '', note: str = '') -> dict:

• main.py:17-36 (helper)

  Database initialization helper that creates the 'expenses' table with all required columns (id, description, amount, category, subcategory, date, note, created_at, updated_at) if it doesn't already exist.

  async def init_db():
      try:
          async with aiosqlite.connect(DB_PATH) as conn:
              await conn.execute("""
                  CREATE TABLE IF NOT EXISTS expenses (
                      id INTEGER PRIMARY KEY AUTOINCREMENT,
                      description TEXT NOT NULL,
                      amount REAL NOT NULL,
                      category TEXT NOT NULL,
                      subcategory TEXT DEFAULT '',
                      date TEXT NOT NULL,
                      note TEXT DEFAULT '',
                      created_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP,
                      updated_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP
                  )
              """)
              await conn.commit()
      except aiosqlite.Error as e:
          print(f"Error initializing database: {e}")
          raise