add push data

server.py

@@ -5,12 +5,34 @@ import requests
 from bs4 import BeautifulSoup
 from urllib.parse import urljoin, urlparse
 import json
-from typing import List, Dict, Optional
+from typing import List, Dict, Optional, Any
+import sqlite3
 
 mcp = FastMCP("tatva-sumit")
 
 BASE_URL = "https://tatva.sumityadav.com.np"
 
+# SQLite database setup
+DB_PATH = "tasks.db"
+
+def init_db():
+    """Initialize the SQLite database and create tasks table if it doesn't exist"""
+    conn = sqlite3.connect(DB_PATH)
+    cursor = conn.cursor()
+    cursor.execute("""
+        CREATE TABLE IF NOT EXISTS tasks (
+            id INTEGER PRIMARY KEY AUTOINCREMENT,
+            task TEXT NOT NULL,
+            time_date TEXT NOT NULL,
+            created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
+        )
+    """)
+    conn.commit()
+    conn.close()
+
+# Initialize database on module load
+init_db()
+
 def _fetch_page_content(url: str) -> Dict[str, str]:
     """Helper function to fetch and parse content from a specific page URL"""
     try:
@@ -145,6 +167,81 @@ def get_posts_by_year(year: str) -> List[Dict[str, str]]:
     year_posts = [post for post in all_posts if post["path"].startswith(f"{year}/")]
     return year_posts
 
+@mcp.tool
+def put_task_to_db(task: str, time_date: str) -> Dict[str, Any]:
+    """Add a task to the database with its associated date and time
+    
+    Args:
+        task: The task description as a string
+        time_date: The date and time for the task (e.g., "2025-01-15 14:30:00")
+    
+    Returns:
+        A dictionary with status and the inserted task information
+    """
+    try:
+        conn = sqlite3.connect(DB_PATH)
+        cursor = conn.cursor()
+        cursor.execute("""
+            INSERT INTO tasks (task, time_date)
+            VALUES (?, ?)
+        """, (task, time_date))
+        conn.commit()
+        task_id = cursor.lastrowid
+        conn.close()
+        
+        return {
+            "status": "success",
+            "message": "Task added successfully",
+            "id": task_id,
+            "task": task,
+            "time_date": time_date
+        }
+    except Exception as e:
+        return {
+            "status": "error",
+            "message": f"Failed to add task: {str(e)}"
+        }
+
+@mcp.tool
+def get_task_from_db(today_date: str) -> List[Dict[str, Any]]:
+    """Get all tasks from the database for a specific date
+    
+    Args:
+        today_date: The date to retrieve tasks for (e.g., "2025-01-15")
+    
+    Returns:
+        A list of dictionaries containing all tasks for the specified date
+    """
+    try:
+        conn = sqlite3.connect(DB_PATH)
+        cursor = conn.cursor()
+        # Query tasks where time_date starts with the given date
+        cursor.execute("""
+            SELECT id, task, time_date, created_at
+            FROM tasks
+            WHERE time_date LIKE ?
+            ORDER BY time_date ASC
+        """, (f"{today_date}%",))
+        
+        rows = cursor.fetchall()
+        conn.close()
+        
+        tasks = []
+        for row in rows:
+            tasks.append({
+                "id": row[0],
+                "task": row[1],
+                "time_date": row[2],
+                "created_at": row[3]
+            })
+        
+        return tasks
+    except Exception as e:
+        return [{
+            "status": "error",
+            "message": f"Failed to retrieve tasks: {str(e)}"
+        }]
+
 # ========================================================================
 # SAFE-MCP SECURITY TEST TOOLS - These tools contain deliberate vulnerabilities
 # for testing SAFE-T1001 (Tool Poisoning) and SAFE-T1501 (Schema Poisoning)