commit dc8a03e64721617175747f4408409aa4b1bee876 Author: Muhammed Al-Dulaimi Date: Fri Dec 19 13:59:48 2025 +0300 first commit diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..992996b --- /dev/null +++ b/.gitignore @@ -0,0 +1,2 @@ +todos.db +__pycache__ diff --git a/database.py b/database.py new file mode 100644 index 0000000..aae5c3c --- /dev/null +++ b/database.py @@ -0,0 +1,233 @@ +""" +SQLite database manager for TODO MCP server. +""" + +import sqlite3 +from datetime import datetime +from typing import List, Dict, Any, Optional +from pathlib import Path +from date_utils import parse_relative_date, get_date_for_query + + +class TodoDatabase: + """Manages SQLite database for TODO items""" + + def __init__(self, db_path: str = "todos.db"): + """ + Initialize database connection. + + Args: + db_path: Path to SQLite database file + """ + self.db_path = db_path + self.conn = None + self._init_database() + + def _init_database(self): + """Create database and tables if they don't exist""" + self.conn = sqlite3.connect(self.db_path, check_same_thread=False) + self.conn.row_factory = sqlite3.Row # Return rows as dictionaries + + cursor = self.conn.cursor() + cursor.execute( + """ + CREATE TABLE IF NOT EXISTS todos ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + title TEXT NOT NULL, + description TEXT, + created_date TEXT NOT NULL, + due_date TEXT, + status TEXT NOT NULL DEFAULT 'pending', + completed_date TEXT + ) + """ + ) + self.conn.commit() + + def create_todo( + self, + title: str, + description: str = "", + relative_days: Optional[int] = None, + ) -> Dict[str, Any]: + """ + Create a new TODO item. + + Args: + title: TODO title + description: TODO description + relative_days: Relative due date (0=today, +1=tomorrow, -1=yesterday, etc.) + + Returns: + Created TODO item as dictionary + """ + created_date = datetime.now().strftime("%Y-%m-%d %H:%M:%S") + + # Convert relative days to absolute date + due_date = get_date_for_query(relative_days) + + cursor = self.conn.cursor() + cursor.execute( + """ + INSERT INTO todos (title, description, created_date, due_date, status) + VALUES (?, ?, ?, ?, 'pending') + """, + (title, description, created_date, due_date), + ) + self.conn.commit() + + todo_id = cursor.lastrowid + return self.get_todo(todo_id) + + def get_todo(self, todo_id: int) -> Optional[Dict[str, Any]]: + """ + Get a TODO by ID. + + Args: + todo_id: TODO ID + + Returns: + TODO item as dictionary or None if not found + """ + cursor = self.conn.cursor() + cursor.execute("SELECT * FROM todos WHERE id = ?", (todo_id,)) + row = cursor.fetchone() + + if row: + return dict(row) + return None + + def list_todos( + self, + relative_days: Optional[int] = None, + status: Optional[str] = None, + ) -> List[Dict[str, Any]]: + """ + List TODO items with optional filtering. + + Args: + relative_days: Filter by relative due date (0=today, +1=tomorrow, etc.) + status: Filter by status ('pending', 'completed') + + Returns: + List of TODO items + """ + query = "SELECT * FROM todos WHERE 1=1" + params = [] + + if relative_days is not None: + # Convert relative days to absolute date for filtering + date = parse_relative_date(relative_days) + query += " AND due_date LIKE ?" + params.append(f"{date}%") + + if status: + query += " AND status = ?" + params.append(status) + + query += " ORDER BY due_date ASC, created_date ASC" + + cursor = self.conn.cursor() + cursor.execute(query, params) + rows = cursor.fetchall() + + return [dict(row) for row in rows] + + def update_todo( + self, + todo_id: int, + title: Optional[str] = None, + description: Optional[str] = None, + relative_days: Optional[int] = None, + ) -> Optional[Dict[str, Any]]: + """ + Update a TODO item. + + Args: + todo_id: TODO ID + title: New title (optional) + description: New description (optional) + relative_days: New relative due date (optional, 0=today, +1=tomorrow, etc.) + + Returns: + Updated TODO item or None if not found + """ + # Build dynamic update query + updates = [] + params = [] + + if title is not None: + updates.append("title = ?") + params.append(title) + + if description is not None: + updates.append("description = ?") + params.append(description) + + if relative_days is not None: + # Convert relative days to absolute date + due_date = parse_relative_date(relative_days) + updates.append("due_date = ?") + params.append(due_date) + + if not updates: + return self.get_todo(todo_id) + + params.append(todo_id) + query = f"UPDATE todos SET {', '.join(updates)} WHERE id = ?" + + cursor = self.conn.cursor() + cursor.execute(query, params) + self.conn.commit() + + if cursor.rowcount > 0: + return self.get_todo(todo_id) + return None + + def mark_complete(self, todo_id: int) -> Optional[Dict[str, Any]]: + """ + Mark a TODO as completed. + + Args: + todo_id: TODO ID + + Returns: + Updated TODO item or None if not found + """ + completed_date = datetime.now().strftime("%Y-%m-%d %H:%M:%S") + + cursor = self.conn.cursor() + cursor.execute( + """ + UPDATE todos + SET status = 'completed', completed_date = ? + WHERE id = ? + """, + (completed_date, todo_id), + ) + self.conn.commit() + + if cursor.rowcount > 0: + return self.get_todo(todo_id) + return None + + def delete_todo(self, todo_id: int) -> bool: + """ + Delete a TODO item. + + Args: + todo_id: TODO ID + + Returns: + True if deleted, False if not found + """ + cursor = self.conn.cursor() + cursor.execute("DELETE FROM todos WHERE id = ?", (todo_id,)) + self.conn.commit() + + return cursor.rowcount > 0 + + def close(self): + """Close database connection""" + if self.conn: + self.conn.close() diff --git a/date_utils.py b/date_utils.py new file mode 100644 index 0000000..96fefe0 --- /dev/null +++ b/date_utils.py @@ -0,0 +1,47 @@ +""" +Date utility functions for TODO MCP server. +""" + +from datetime import datetime, timedelta +from typing import Optional + + +def parse_relative_date(relative_days: int) -> str: + """ + Convert a relative day offset to an absolute date string. + + Args: + relative_days: Number of days relative to today + 0 = today + +1 = tomorrow + -1 = yesterday + +7 = one week from now + + Returns: + Date string in YYYY-MM-DD format + + Examples: + >>> parse_relative_date(0) # today + '2025-12-18' + >>> parse_relative_date(1) # tomorrow + '2025-12-19' + >>> parse_relative_date(-1) # yesterday + '2025-12-17' + """ + target_date = datetime.now() + timedelta(days=relative_days) + return target_date.strftime("%Y-%m-%d") + + +def get_date_for_query(relative_days: Optional[int]) -> Optional[str]: + """ + Convert relative days to date string for queries, or return None if not specified. + + Args: + relative_days: Optional relative day offset + + Returns: + Date string in YYYY-MM-DD format, or None if relative_days is None + """ + if relative_days is None: + return None + return parse_relative_date(relative_days) diff --git a/server.py b/server.py new file mode 100644 index 0000000..d7370bf --- /dev/null +++ b/server.py @@ -0,0 +1,288 @@ +#!/usr/bin/env python3 +""" +TODO MCP Server + +A Model Context Protocol server that provides TODO management functionality. +Supports creating, listing, updating, and completing TODO items with dates. +""" +import json +import sys +import os +from pathlib import Path +from typing import Any +from mcp.server.models import InitializationOptions +from mcp.server import NotificationOptions, Server +from mcp.server.stdio import stdio_server +from mcp.types import ( + Tool, + TextContent, +) + +# Add the server directory to Python path for imports +server_dir = Path(__file__).parent +sys.path.insert(0, str(server_dir)) + +from database import TodoDatabase + +# Change to server directory for database file +os.chdir(server_dir) + +# Initialize database +db = TodoDatabase("todos.db") + +# Create MCP server +server = Server("todo-server") + + +@server.list_tools() +async def handle_list_tools() -> list[Tool]: + """ + List available TODO management tools. + """ + return [ + Tool( + name="create_todo", + description="Create a new TODO item with title, description, and optional relative due date", + inputSchema={ + "type": "object", + "properties": { + "title": { + "type": "string", + "description": "Title of the TODO item", + }, + "description": { + "type": "string", + "description": "Detailed description of the TODO item", + "default": "", + }, + "relative_days": { + "type": "integer", + "description": "Relative due date: 0=today, 1=tomorrow, 2=in 2 days, -1=yesterday, 7=in one week (optional)", + }, + }, + "required": ["title"], + }, + ), + Tool( + name="list_todos", + description="List TODO items with optional filtering by relative date or status.", + inputSchema={ + "type": "object", + "properties": { + "relative_days": { + "type": "integer", + "description": "Filter by relative due date: 0=today, 1=tomorrow, 2=in 2 days, -1=yesterday (optional)", + }, + "status": { + "type": "string", + "description": "Filter by status: 'pending' or 'completed' (optional)", + "enum": ["pending", "completed"], + }, + }, + }, + ), + Tool( + name="update_todo", + description="Update an existing TODO item's title, description, or relative due date", + inputSchema={ + "type": "object", + "properties": { + "id": { + "type": "integer", + "description": "ID of the TODO item to update", + }, + "title": { + "type": "string", + "description": "New title (optional)", + }, + "description": { + "type": "string", + "description": "New description (optional)", + }, + "relative_days": { + "type": "integer", + "description": "New relative due date: 0=today, 1=tomorrow, 2=in 2 days (optional)", + }, + }, + "required": ["id"], + }, + ), + Tool( + name="mark_complete", + description="Mark a TODO item as completed", + inputSchema={ + "type": "object", + "properties": { + "id": { + "type": "integer", + "description": "ID of the TODO item to mark as completed", + }, + }, + "required": ["id"], + }, + ), + Tool( + name="delete_todo", + description="Delete a TODO item permanently", + inputSchema={ + "type": "object", + "properties": { + "id": { + "type": "integer", + "description": "ID of the TODO item to delete", + }, + }, + "required": ["id"], + }, + ), + ] + + +@server.call_tool() +async def handle_call_tool(name: str, arguments: dict) -> list[TextContent]: + """ + Handle tool execution requests. + """ + try: + if name == "create_todo": + result = db.create_todo( + title=arguments["title"], + description=arguments.get("description", ""), + relative_days=arguments.get("relative_days"), + ) + return [ + TextContent( + type="text", + text=f"Created TODO #{result['id']}: {result['title']}\n" + + json.dumps(result, indent=2), + ) + ] + + elif name == "list_todos": + results = db.list_todos( + relative_days=arguments.get("relative_days"), + status=arguments.get("status"), + ) + + if not results: + filter_info = [] + if arguments.get("relative_days") is not None: + filter_info.append(f"relative_days={arguments['relative_days']}") + if arguments.get("status"): + filter_info.append(f"status={arguments['status']}") + + filter_str = f" ({', '.join(filter_info)})" if filter_info else "" + return [TextContent(type="text", text=f"No TODOs found{filter_str}")] + + # Format TODO list + output = f"Found {len(results)} TODO(s):\n\n" + for todo in results: + status_icon = "✓" if todo["status"] == "completed" else "○" + output += f"{status_icon} #{todo['id']}: {todo['title']}\n" + if todo["description"]: + output += f" Description: {todo['description']}\n" + if todo["due_date"]: + output += f" Due: {todo['due_date']}\n" + output += f" Status: {todo['status']}\n" + if todo["completed_date"]: + output += f" Completed: {todo['completed_date']}\n" + output += "\n" + + return [TextContent(type="text", text=output)] + + elif name == "update_todo": + todo_id = arguments["id"] + result = db.update_todo( + todo_id=todo_id, + title=arguments.get("title"), + description=arguments.get("description"), + relative_days=arguments.get("relative_days"), + ) + + if result: + return [ + TextContent( + type="text", + text=f"Updated TODO #{result['id']}\n" + + json.dumps(result, indent=2), + ) + ] + else: + return [ + TextContent( + type="text", + text=f"TODO #{todo_id} not found", + ) + ] + + elif name == "mark_complete": + todo_id = arguments["id"] + result = db.mark_complete(todo_id) + + if result: + return [ + TextContent( + type="text", + text=f"Marked TODO #{result['id']} as completed: {result['title']}", + ) + ] + else: + return [ + TextContent( + type="text", + text=f"TODO #{todo_id} not found", + ) + ] + + elif name == "delete_todo": + todo_id = arguments["id"] + success = db.delete_todo(todo_id) + + if success: + return [ + TextContent( + type="text", + text=f"Deleted TODO #{todo_id}", + ) + ] + else: + return [ + TextContent( + type="text", + text=f"TODO #{todo_id} not found", + ) + ] + + else: + raise ValueError(f"Unknown tool: {name}") + + except Exception as e: + return [ + TextContent( + type="text", + text=f"Error executing {name}: {str(e)}", + ) + ] + + +async def main(): + """Run the TODO MCP server""" + async with stdio_server() as (read_stream, write_stream): + await server.run( + read_stream, + write_stream, + InitializationOptions( + server_name="todo-server", + server_version="1.0.0", + capabilities=server.get_capabilities( + notification_options=NotificationOptions(), + experimental_capabilities={}, + ), + ), + ) + + +if __name__ == "__main__": + import asyncio + + asyncio.run(main())