هذا الالتزام موجود في:
2025-12-19 13:59:48 +03:00
التزام dc8a03e647
4 ملفات معدلة مع 570 إضافات و0 حذوفات

2
.gitignore مباع Normal file
عرض الملف

@@ -0,0 +1,2 @@
todos.db
__pycache__

233
database.py Normal file
عرض الملف

@@ -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()

47
date_utils.py Normal file
عرض الملف

@@ -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)

288
server.py Normal file
عرض الملف

@@ -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())