289 أسطر
9.3 KiB
Python
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())
|