الملفات
todo-mcp-server/server.py
2025-12-19 13:59:48 +03:00

289 أسطر
9.3 KiB
Python

#!/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())