Implement task notification system with cron job and ntfy integration

.env.example

@@ -0,0 +1,24 @@
+# Rowmind Environment Variables
+# Copy this file to .env and fill in your values
+
+# ===================
+# ntfy Configuration
+# ===================
+# Server URL (default: https://ntfy.sh, or use your self-hosted instance)
+NTFY_SERVER=https://ntfy.sh
+
+# Topic name - USE A RANDOM/UNGUESSABLE NAME for security!
+# Example: rowmind-a3f8b2c1-notifications
+# Generate one with: python -c "import uuid; print(f'rowmind-{uuid.uuid4()}')"
+NTFY_TOPIC=rowmind-your-random-topic-here
+
+# Optional: Access token for authentication (recommended for public repos)
+# Create one at https://ntfy.sh/account or your self-hosted instance
+# This prevents others from sending notifications to your topic
+NTFY_TOKEN=
+
+# ===================
+# Cron Configuration
+# ===================
+# How often to check for due tasks (in seconds)
+CRON_INTERVAL_SECONDS=60

.gitignore

@@ -1,4 +1,7 @@
 data/*
 plan.md
 __pycache__/
-*.log
+*.log
+
+# Environment variables (contains secrets)
+.env

README.md

@@ -20,6 +20,43 @@ tags:
 
 ## Rowmind MCP Server
 
+### 🧠 How it Works
+
+```mermaid
+graph LR
+    User[User / Chat Client] <-->|MCP Protocol| Server[Rowmind Server]
+    Server <-->|Reads/Writes| CSVs[(Local CSV Files)]
+    
+    subgraph "Your Machine / Private Space"
+    Server
+    CSVs
+    end
+    
+    style User fill:#f9f,stroke:#333,stroke-width:2px
+    style Server fill:#bbf,stroke:#333,stroke-width:2px
+    style CSVs fill:#bfb,stroke:#333,stroke-width:2px
+```
+
+### 🚀 Why Rowmind?
+
+```mermaid
+mindmap
+  root((Rowmind))
+    Privacy
+      Data lives locally
+      You own your memory
+      No external DB
+    Simplicity
+      Plain text CSVs
+      Easy to backup
+      Easy to edit
+    Control
+      Custom Tables
+      Your own Schema
+      Hackable
+```
+
+
 ### Idea
 
 Rowmind is a small local “memory service” that you plug into any MCP‑compatible chat client (ChatGPT, Claude, Cursor, etc.).
@@ -80,15 +117,82 @@ Rowmind exposes a small set of tools and one resource over MCP:
 
 The same pattern works for expenses, mood logs, trips, or any other table you define.
 
-### Reminders
+### Reminders & Notifications 🔔
+
+> **🧪 Hackathon Experiment**: This feature explores how an MCP server can go beyond just storing data—it can actively reach out to you when something needs attention. Instead of you asking "do I have any pending tasks?", Rowmind tells you.
+
+#### Why ntfy?
+
+We wanted the simplest possible way to send push notifications without building a mobile app or dealing with Firebase/APNs complexity. [ntfy](https://ntfy.sh) is perfect for this:
+
+- **No account required** — just pick a topic name and subscribe
+- **Works everywhere** — Android, iOS, web, CLI
+- **Self-hostable** — if you need full control
+- **HTTP-based** — a simple POST request sends a notification
+
+This makes ntfy ideal for hackathon projects where you want real push notifications without the usual infrastructure overhead.
+
+#### How it works
+
+Rowmind runs a **background worker** (using APScheduler) that:
+
+1. Checks every 60 seconds for tasks where `scheduled_at_epoch <= now`
+2. Sends a push notification via ntfy with the task description
+3. Marks one-time tasks as `done`, or reschedules recurring tasks
+
+```text
+┌─────────────┐     ┌─────────────┐     ┌─────────────┐
+│  tasks.csv  │ --> │ Cron Worker │ --> │    ntfy     │ --> 📱 Your phone
+│  (pending)  │     │ (every 60s) │     │   (HTTP)    │
+└─────────────┘     └─────────────┘     └─────────────┘
+```
+
+#### Quick setup
+
+1. **Copy the environment template:**
+
+   ```bash
+   cp .env.example .env
+   ```
+
+2. **Generate a random topic name** (important for security):
+
+   ```bash
+   python -c "import uuid; print(f'rowmind-{uuid.uuid4()}')"
+   ```
+
+3. **Configure your `.env` file:**
+
+   ```env
+   # Required: Use a random topic name for security
+   NTFY_TOPIC=rowmind-your-random-topic-here
+   
+   # Optional: ntfy server (defaults to https://ntfy.sh)
+   NTFY_SERVER=https://ntfy.sh
+   
+   # Optional but recommended: Access token for authentication
+   NTFY_TOKEN=tk_your_access_token
+   
+   # Optional: Check interval in seconds (default: 60)
+   CRON_INTERVAL_SECONDS=60
+   ```
+
+4. **Subscribe to your topic** in the ntfy app ([Android](https://play.google.com/store/apps/details?id=io.heckel.ntfy), [iOS](https://apps.apple.com/app/ntfy/id1625396347), or [web](https://ntfy.sh))
+
+5. **Run the app** — the scheduler starts automatically:
+
+   ```bash
+   python app.py
+   ```
+
+#### Security considerations
 
-A simple worker script can turn tasks into notifications:
+Since this is a public repository:
 
-- Read `tasks.csv`.
-- Compute `now_epoch`.
-- Find tasks with `status="pending"` and `scheduled_at_epoch <= now_epoch`.
-- Send a notification (for example via ntfy).
-- If `schedule_type="one_time"` or `interval_seconds=0`, mark the task as `done`; otherwise add `interval_seconds` to `scheduled_at_epoch` and keep it `pending`.
+- **Never commit your `.env` file** (it's in `.gitignore`)
+- Use a **random/unguessable topic name** — anyone who knows your topic can send you notifications
+- Consider using **ntfy access tokens** for authentication
+- For production use, run your own [self-hosted ntfy server](https://docs.ntfy.sh/install/)
 
 Because everything is stored as epoch timestamps, the worker only does integer comparisons and additions.
 

app.py

@@ -2,6 +2,9 @@ import os
 from typing import List, Dict
 
 import gradio as gr
+from dotenv import load_dotenv
+
+from cron import start_scheduler, is_scheduler_running
 from helper import (
     get_csv_path,
     get_current_time_epoch,
@@ -411,5 +414,12 @@ with gr.Blocks(title="Rowmind MCP Server") as demo:
                         api_name="update_row",
                     )
 
+# Load environment variables
+load_dotenv()
+
 if __name__ == "__main__":
+    # Start the background cron scheduler for task notifications
+    if not is_scheduler_running():
+        start_scheduler()
+
     demo.launch(mcp_server=True)

cron.py

@@ -0,0 +1,264 @@
+"""
+Cron worker for Rowmind task notifications.
+
+This module provides a background scheduler that:
+1. Checks for tasks with scheduled_at_epoch <= current time
+2. Sends notifications via ntfy
+3. Updates task status (done for one-time, recalculates next run for recurring)
+
+Uses APScheduler as recommended by Gradio for background tasks.
+"""
+
+import logging
+import os
+import time
+from typing import Optional
+
+import httpx
+from dotenv import load_dotenv
+from apscheduler.schedulers.background import BackgroundScheduler
+
+from helper import (
+    execute_sql_query,
+    get_csv_path,
+    update_row_in_csv,
+)
+
+# Load environment variables
+load_dotenv()
+
+# Configure logging
+logging.basicConfig(level=logging.INFO)
+logger = logging.getLogger(__name__)
+
+# ntfy Configuration
+NTFY_SERVER = os.getenv("NTFY_SERVER", "https://ntfy.sh")
+NTFY_TOPIC = os.getenv("NTFY_TOPIC")
+NTFY_TOKEN = os.getenv("NTFY_TOKEN")  # Optional, for authentication
+
+# Cron Configuration
+CRON_INTERVAL_SECONDS = int(os.getenv("CRON_INTERVAL_SECONDS", "60"))
+
+# Global scheduler instance
+_scheduler: Optional[BackgroundScheduler] = None
+
+
+def send_ntfy_notification(
+    message: str, priority: str = "default", tags: str = "bell"
+) -> bool:
+    """
+    Send a notification via ntfy.
+
+    Args:
+        message: The notification message body
+        priority: Priority level (min, low, default, high, urgent)
+        tags: Comma-separated emoji tags (e.g., "bell,warning")
+
+    Returns:
+        True if notification was sent successfully, False otherwise
+    """
+    if not NTFY_TOPIC:
+        logger.warning("NTFY_TOPIC not configured. Skipping notification.")
+        return False
+
+    url = f"{NTFY_SERVER.rstrip('/')}/{NTFY_TOPIC}"
+
+    headers = {
+        "Title": "Rowmind Task",
+        "Priority": priority,
+        "Tags": tags,
+    }
+
+    # Add authentication if token is configured
+    if NTFY_TOKEN:
+        headers["Authorization"] = f"Bearer {NTFY_TOKEN}"
+
+    try:
+        response = httpx.post(
+            url,
+            content=message.encode("utf-8"),
+            headers=headers,
+            timeout=10.0,
+        )
+        response.raise_for_status()
+        logger.info("Notification sent successfully")
+        return True
+    except httpx.HTTPStatusError as e:
+        logger.error(
+            f"Failed to send notification (HTTP {e.response.status_code}): {e}"
+        )
+        return False
+    except httpx.RequestError as e:
+        logger.error(f"Failed to send notification (network error): {e}")
+        return False
+
+
+def get_due_tasks() -> list[dict]:
+    """
+    Get all pending tasks that are due for notification.
+
+    Returns:
+        List of task dictionaries with rowid included
+    """
+    now = int(time.time())
+
+    query = f"""
+        SELECT *, rowid
+        FROM tasks
+        WHERE status = 'pending'
+        AND scheduled_at_epoch > 0
+        AND scheduled_at_epoch <= {now}
+        ORDER BY scheduled_at_epoch ASC
+    """
+
+    result = execute_sql_query(query)
+
+    # Filter out error responses
+    if result and "error" in result[0]:
+        logger.error(f"SQL error getting due tasks: {result[0]['error']}")
+        return []
+
+    return result
+
+
+def process_task(task: dict) -> None:
+    """
+    Process a due task: send notification and update status.
+
+    Args:
+        task: Task dictionary from the database
+    """
+    description = task.get("description", "Task reminder")
+    task_id = task.get("task_id", "unknown")
+    schedule_type = task.get("schedule_type", "one_time")
+    interval_seconds = int(task.get("interval_seconds", 0) or 0)
+    rowid = task.get("rowid")
+
+    logger.info(f"Processing task {task_id}: {description}")
+
+    # Send notification
+    if not send_ntfy_notification(message=description, tags="memo,bell"):
+        logger.warning(f"Failed to notify for task {task_id}, will retry next cycle")
+        return
+
+    # Update task status
+    tasks_csv = get_csv_path("tasks")
+
+    if schedule_type == "recurring" and interval_seconds > 0:
+        # Recurring task: update next scheduled time
+        current_scheduled = int(task.get("scheduled_at_epoch", 0) or 0)
+        next_scheduled = current_scheduled + interval_seconds
+
+        update_row_in_csv(
+            tasks_csv,
+            key_field="rowid",
+            key_value=rowid,
+            update_data={"scheduled_at_epoch": str(next_scheduled)},
+        )
+        logger.info(f"Task {task_id} rescheduled for epoch {next_scheduled}")
+    else:
+        # One-time task: mark as done
+        update_row_in_csv(
+            tasks_csv,
+            key_field="rowid",
+            key_value=rowid,
+            update_data={"status": "done"},
+        )
+        logger.info(f"Task {task_id} marked as done")
+
+
+def run_cron_job() -> None:
+    """
+    Main cron job function. Checks for due tasks and processes them.
+    """
+    logger.debug("Running cron job...")
+
+    due_tasks = get_due_tasks()
+
+    if not due_tasks:
+        logger.debug("No due tasks found")
+        return
+
+    logger.info(f"Found {len(due_tasks)} due task(s)")
+
+    for task in due_tasks:
+        try:
+            process_task(task)
+        except Exception as e:
+            logger.error(f"Error processing task {task.get('task_id', 'unknown')}: {e}")
+
+
+def start_scheduler() -> BackgroundScheduler:
+    """
+    Start the background scheduler for the cron job.
+
+    Returns:
+        The started BackgroundScheduler instance
+    """
+    global _scheduler
+
+    if _scheduler is not None and _scheduler.running:
+        logger.warning("Scheduler already running")
+        return _scheduler
+
+    if not NTFY_TOPIC:
+        logger.warning(
+            "NTFY_TOPIC not configured. Set it in .env to enable notifications. "
+            "Scheduler will still run but notifications won't be sent."
+        )
+
+    _scheduler = BackgroundScheduler()
+    _scheduler.add_job(
+        func=run_cron_job,
+        trigger="interval",
+        seconds=CRON_INTERVAL_SECONDS,
+        id="rowmind_task_notifier",
+        replace_existing=True,
+    )
+    _scheduler.start()
+
+    logger.info(
+        f"Cron scheduler started. Checking for due tasks every {CRON_INTERVAL_SECONDS} seconds."
+    )
+
+    return _scheduler
+
+
+def stop_scheduler() -> None:
+    """Stop the background scheduler if running."""
+    global _scheduler
+
+    if _scheduler is not None and _scheduler.running:
+        _scheduler.shutdown(wait=False)
+        logger.info("Cron scheduler stopped")
+
+    _scheduler = None
+
+
+def is_scheduler_running() -> bool:
+    """Check if the scheduler is currently running."""
+    return _scheduler is not None and _scheduler.running
+
+
+# For standalone testing
+if __name__ == "__main__":
+    import time
+
+    print("Starting cron worker in standalone mode...")
+    print(f"NTFY_SERVER: {NTFY_SERVER}")
+    print(f"NTFY_TOPIC: {NTFY_TOPIC or '(not configured)'}")
+    print(f"CRON_INTERVAL_SECONDS: {CRON_INTERVAL_SECONDS}")
+
+    scheduler = start_scheduler()
+
+    try:
+        # Run immediately once for testing
+        run_cron_job()
+
+        # Keep the main thread alive
+        while True:
+            time.sleep(1)
+    except KeyboardInterrupt:
+        print("\nStopping scheduler...")
+        stop_scheduler()
+        print("Done.")

pyproject.toml

@@ -5,6 +5,7 @@ description = "MCP server"
 readme = "README.md"
 requires-python = ">=3.11"
 dependencies = [
+    "apscheduler>=3.11.0",
     "duckdb>=1.4.2",
     "gradio[mcp,oauth]>=5.49.1",
     "huggingface-hub>=1.1.4",

requirements.txt

@@ -14,6 +14,8 @@ anyio==4.11.0
     #   openai
     #   sse-starlette
     #   starlette
+apscheduler==3.11.1
+    # via rowmind (pyproject.toml)
 attrs==25.4.0
     # via
     #   jsonschema
@@ -260,6 +262,8 @@ typing-inspection==0.4.2
     #   pydantic-settings
 tzdata==2025.2
     # via pandas
+tzlocal==5.3.1
+    # via apscheduler
 urllib3==2.5.0
     # via requests
 uvicorn==0.38.0

uv.lock

@@ -48,6 +48,18 @@ wheels = [
     { url = "https://files.pythonhosted.org/packages/15/b3/9b1a8074496371342ec1e796a96f99c82c945a339cd81a8e73de28b4cf9e/anyio-4.11.0-py3-none-any.whl", hash = "sha256:0287e96f4d26d4149305414d4e3bc32f0dcd0862365a4bddea19d7a1ec38c4fc", size = 109097, upload-time = "2025-09-23T09:19:10.601Z" },
 ]
 
+[[package]]
+name = "apscheduler"
+version = "3.11.1"
+source = { registry = "https://pypi.org/simple" }
+dependencies = [
+    { name = "tzlocal" },
+]
+sdist = { url = "https://files.pythonhosted.org/packages/d0/81/192db4f8471de5bc1f0d098783decffb1e6e69c4f8b4bc6711094691950b/apscheduler-3.11.1.tar.gz", hash = "sha256:0db77af6400c84d1747fe98a04b8b58f0080c77d11d338c4f507a9752880f221", size = 108044, upload-time = "2025-10-31T18:55:42.819Z" }
+wheels = [
+    { url = "https://files.pythonhosted.org/packages/58/9f/d3c76f76c73fcc959d28e9def45b8b1cc3d7722660c5003b19c1022fd7f4/apscheduler-3.11.1-py3-none-any.whl", hash = "sha256:6162cb5683cb09923654fa9bdd3130c4be4bfda6ad8990971c9597ecd52965d2", size = 64278, upload-time = "2025-10-31T18:55:41.186Z" },
+]
+
 [[package]]
 name = "attrs"
 version = "25.4.0"
@@ -1727,6 +1739,7 @@ name = "rowmind"
 version = "0.1.0"
 source = { virtual = "." }
 dependencies = [
+    { name = "apscheduler" },
     { name = "duckdb" },
     { name = "gradio", extra = ["mcp", "oauth"] },
     { name = "huggingface-hub" },
@@ -1736,6 +1749,7 @@ dependencies = [
 
 [package.metadata]
 requires-dist = [
+    { name = "apscheduler", specifier = ">=3.11.0" },
     { name = "duckdb", specifier = ">=1.4.2" },
     { name = "gradio", extras = ["mcp", "oauth"], specifier = ">=5.49.1" },
     { name = "huggingface-hub", specifier = ">=1.1.4" },
@@ -2070,6 +2084,18 @@ wheels = [
     { url = "https://files.pythonhosted.org/packages/5c/23/c7abc0ca0a1526a0774eca151daeb8de62ec457e77262b66b359c3c7679e/tzdata-2025.2-py2.py3-none-any.whl", hash = "sha256:1a403fada01ff9221ca8044d701868fa132215d84beb92242d9acd2147f667a8", size = 347839, upload-time = "2025-03-23T13:54:41.845Z" },
 ]
 
+[[package]]
+name = "tzlocal"
+version = "5.3.1"
+source = { registry = "https://pypi.org/simple" }
+dependencies = [
+    { name = "tzdata", marker = "sys_platform == 'win32'" },
+]
+sdist = { url = "https://files.pythonhosted.org/packages/8b/2e/c14812d3d4d9cd1773c6be938f89e5735a1f11a9f184ac3639b93cef35d5/tzlocal-5.3.1.tar.gz", hash = "sha256:cceffc7edecefea1f595541dbd6e990cb1ea3d19bf01b2809f362a03dd7921fd", size = 30761, upload-time = "2025-03-05T21:17:41.549Z" }
+wheels = [
+    { url = "https://files.pythonhosted.org/packages/c2/14/e2a54fabd4f08cd7af1c07030603c3356b74da07f7cc056e600436edfa17/tzlocal-5.3.1-py3-none-any.whl", hash = "sha256:eb1a66c3ef5847adf7a834f1be0800581b683b5608e74f86ecbcef8ab91bb85d", size = 18026, upload-time = "2025-03-05T21:17:39.857Z" },
+]
+
 [[package]]
 name = "urllib3"
 version = "2.5.0"