Todo MCP Server

A comprehensive Model Context Protocol (MCP) server for AI agent task management, providing structured tools for creating, managing, and organizing tasks through markdown-based storage with advanced features for enterprise-grade task management.

🚀 Features

Core Task Management

• Full CRUD Operations: Create, read, update, and delete tasks with validation
• Hierarchical Organization: Multi-level parent-child task relationships
• Status Tracking: Complete lifecycle management (pending, in-progress, completed, blocked)
• Priority System: Four-level priority system (low, medium, high, urgent)
• Tag-based Categorization: Flexible tagging system for advanced organization
• Due Date Management: Full datetime support with timezone handling

Advanced Capabilities

• Markdown Storage: Human-readable task storage with YAML frontmatter
• MCP Protocol Compliance: Full compatibility with Model Context Protocol v1.0
• HTTP/REST API: Full REST API with OpenAPI documentation
• Server-Sent Events (SSE): Real-time updates and streaming responses
• WebSocket Support: Bidirectional real-time communication
• Streaming Operations: Progressive task operations with live progress updates
• Concurrent Access: Thread-safe multi-agent support
• Performance Optimized: Handles 1000+ tasks with sub-100ms response times
• Caching System: LRU cache with configurable size limits
• Search & Filtering: Full-text search with advanced filtering options
• Audit Trail: Complete tool call history and change tracking

Enterprise Features

• Backup System: Automatic backup with configurable retention
• File Monitoring: Real-time file system change detection
• Configuration Management: Environment-based configuration with validation
• Comprehensive Logging: Structured logging with multiple levels
• Error Recovery: Graceful error handling and system resilience
• Performance Monitoring: Built-in metrics and response time tracking

Installation

Prerequisites

• Python 3.8 or higher
• uv package manager

Setup

1. Clone the repository:

git clone <repository-url>
cd todo-mcp

2. Create virtual environment and install dependencies:

uv sync

3. Install development dependencies:

uv sync --dev

Usage

Running the Servers

MCP Server (stdio protocol)

Start the MCP server with default settings:

uv run todo-mcp-server serve

With custom configuration:

uv run todo-mcp-server serve --data-dir /path/to/data --log-level DEBUG

FastMCP Server (HTTP protocol for AI clients)

Start the FastMCP server for AI client connections:

uv run todo-mcp-server serve-fastmcp

With custom host and port:

uv run todo-mcp-server serve-fastmcp --host 0.0.0.0 --port 8000

AI clients can connect to: http://localhost:8000/mcp

HTTP Server (REST API + SSE + WebSocket)

Start the HTTP server with streaming support:

uv run todo-mcp-server serve-http

With custom host and port:

uv run todo-mcp-server serve-http --host 0.0.0.0 --port 8080

Hybrid Mode (Both servers)

Run both MCP and HTTP servers simultaneously:

uv run todo-mcp-server serve-hybrid --port 8000 --mcp-stdio

Server Features by Mode

Feature	MCP Server	FastMCP Server	HTTP Server	Hybrid Mode
MCP Protocol (stdio)	✅	❌	❌	✅
MCP Protocol (HTTP)	❌	✅	❌	❌
AI Client Support	✅	✅	❌	✅
REST API	❌	❌	✅	✅
Server-Sent Events	❌	✅	✅	✅
WebSocket	❌	❌	✅	✅
Streaming Operations	❌	✅	✅
Real-time Dashboard	❌	✅	✅
OpenAPI Docs	❌	✅	✅

🛠️ Available MCP Tools

Task Management Tools

• create_task: Create a new task with full metadata support
• update_task: Update existing task properties with validation
• delete_task: Delete tasks with optional cascade for children
• get_task: Retrieve specific task with complete information
• list_tasks: List tasks with advanced filtering and pagination
• get_task_context: Get task with full context including hierarchy and history

Hierarchy Management Tools

• add_child_task: Establish parent-child relationships
• remove_child_task: Remove child relationships safely
• get_task_hierarchy: Retrieve complete task hierarchy trees
• move_task: Move tasks between parents with validation

Status Management Tools

• update_task_status: Update individual task status with validation
• bulk_status_update: Batch update multiple task statuses
• get_task_status: Get current status of specific tasks
• get_pending_tasks: Retrieve all tasks in pending state
• get_in_progress_tasks: Retrieve all active tasks
• get_blocked_tasks: Retrieve all blocked tasks
• get_completed_tasks: Retrieve all completed tasks

Advanced Query Tools

• search_tasks: Full-text search across task content
• filter_tasks: Advanced filtering by multiple criteria
• get_task_statistics: Comprehensive task metrics and analytics

📊 Performance Specifications

• Response Time: < 100ms for task operations
• Search Performance: < 200ms for 1000+ task datasets
• Throughput: 50+ operations per second
• Memory Usage: < 1KB per task average
• Concurrent Users: Supports multiple AI agents simultaneously
• Dataset Size: Tested with 5000+ tasks

Development

Project Structure

src/todo_mcp/
├── models/          # Data models (Task, TaskStatus, etc.)
├── services/        # Business logic layer
├── storage/         # File I/O and markdown handling
├── tools/           # MCP tool implementations
├── utils/           # Utility functions
├── config.py        # Configuration management
└── server.py        # Main MCP server

🧪 Testing

Run All Tests

uv run pytest

Test with Coverage

uv run pytest --cov=src/todo_mcp --cov-report=html

Test Categories

# Unit tests
uv run pytest tests/test_models/ tests/test_services/ tests/test_tools/

# Integration tests
uv run pytest tests/test_integration/

# Performance tests
uv run pytest tests/test_integration/test_performance_load.py

# MCP compliance tests
uv run pytest tests/test_integration/test_mcp_compliance.py

# Concurrent access tests
uv run pytest tests/test_integration/test_concurrent_access.py

Test Data Generation

# Generate test datasets for load testing
uv run python -c "
from tests.test_integration.test_data_generator import TestDataGenerator
# Use generator to create test data
"

Code Quality

Format code:

uv run black .
uv run isort .

Lint code:

uv run ruff check .

Type checking:

uv run mypy .

⚙️ Configuration

Environment Variables

The server supports comprehensive configuration through environment variables with the TODO_MCP_ prefix:

Core Settings

• TODO_MCP_DATA_DIRECTORY: Data storage directory (default: data)
• TODO_MCP_LOG_LEVEL: Logging level - DEBUG, INFO, WARNING, ERROR (default: INFO)
• TODO_MCP_SERVER_NAME: MCP server identifier (default: todo-mcp-server)
• TODO_MCP_SERVER_VERSION: Server version (default: 1.0.0)

Performance Settings

• TODO_MCP_MAX_CACHE_SIZE: Maximum cache entries (default: 1000)
• TODO_MCP_AUTO_SAVE_INTERVAL: Auto-save interval in seconds (default: 30)
• TODO_MCP_PERFORMANCE_MONITORING: Enable performance monitoring (default: false)

Storage Settings

• TODO_MCP_BACKUP_ENABLED: Enable automatic backups (default: true)
• TODO_MCP_BACKUP_DIRECTORY: Backup storage location (default: backups)
• TODO_MCP_BACKUP_RETENTION_DAYS: Backup retention period (default: 30)
• TODO_MCP_FILE_WATCH_ENABLED: Enable file system monitoring (default: true)

Task Settings

• TODO_MCP_MAX_TASK_TITLE_LENGTH: Maximum title length (default: 200)
• TODO_MCP_MAX_TASK_DESCRIPTION_LENGTH: Maximum description length (default: 10000)
• TODO_MCP_DEFAULT_TASK_PRIORITY: Default priority for new tasks (default: medium)
• TODO_MCP_TASK_FILE_EXTENSION: Task file extension (default: .md)

Configuration File

Create a .env file in the project root:

TODO_MCP_DATA_DIRECTORY=/path/to/tasks
TODO_MCP_LOG_LEVEL=DEBUG
TODO_MCP_BACKUP_ENABLED=true
TODO_MCP_MAX_CACHE_SIZE=2000
TODO_MCP_PERFORMANCE_MONITORING=true

Command Line Options

# Basic usage
uv run todo-mcp-server

# Custom configuration
uv run todo-mcp-server \
  --data-dir /custom/path \
  --log-level DEBUG \
  --cache-size 2000 \
  --performance-monitoring

# Production mode
uv run todo-mcp-server \
  --environment production \
  --log-file /var/log/todo-mcp.log \
  --no-file-watch

📄 Task File Format

Tasks are stored as markdown files with YAML frontmatter for maximum human readability:

---
id: task-001
title: "Implement User Authentication"
description: "Add secure login and registration functionality"
status: in_progress
priority: high
tags:
  - authentication
  - security
  - backend
created_at: 2024-01-15T10:30:00Z
updated_at: 2024-01-16T14:22:00Z
due_date: 2024-01-20T17:00:00Z
parent_id: epic-001
child_ids:
  - task-002
  - task-003
metadata:
  estimated_hours: 16
  complexity: high
  assignee: "development-team"
tool_calls:
  - tool: create_task
    timestamp: 2024-01-15T10:30:00Z
    agent: "project-manager-ai"
  - tool: update_task_status
    timestamp: 2024-01-16T14:22:00Z
    agent: "developer-ai"
---

# Implementation Details

This task involves implementing a complete authentication system with the following components:

## Requirements
- [ ] User registration with email validation
- [ ] Secure password hashing (bcrypt)
- [ ] JWT token-based authentication
- [ ] Password reset functionality
- [ ] Rate limiting for login attempts

## Technical Specifications
- Use OAuth 2.0 for third-party integration
- Implement RBAC (Role-Based Access Control)
- Add comprehensive logging for security events

## Dependencies
- Depends on database schema migration (task-002)
- Blocks user profile management (task-003)

## Notes
- Consider implementing 2FA in future iteration
- Ensure GDPR compliance for user data

File Organization

data/
├── tasks/
│   ├── task-001_implement-user-auth.md
│   ├── task-002_database-migration.md
│   └── epic-001_user-management.md
├── templates/
│   ├── bug-report.md
│   ├── feature-request.md
│   └── epic-template.md
└── backups/
    ├── 2024-01-15/
    └── 2024-01-16/

💡 Usage Examples

MCP Protocol Usage

Stdio Transport (Traditional)

# Example using the MCP Python SDK client
import asyncio
from mcp.client.session import ClientSession
from mcp.client.stdio import StdioServerParameters, stdio_client

async def main():
    server_params = StdioServerParameters(
        command="uv",
        args=["run", "todo-mcp-server", "serve"]
    )
    
    async with stdio_client(server_params) as (read, write):
        async with ClientSession(read, write) as session:
            await session.initialize()
            
            # Create a task
            result = await session.call_tool("create_task", {
                "title": "Complete project documentation",
                "description": "Write comprehensive docs",
                "priority": "high"
            })
            print(f"Created task: {result}")

asyncio.run(main())

HTTP Transport (AI Clients)

# Example using StreamableHttp transport for AI clients
import asyncio
from mcp import ClientSession
from mcp.client.streamable_http import streamablehttp_client

async def main():
    # Connect to FastMCP server
    async with streamablehttp_client("http://localhost:8000/mcp") as (read, write, _):
        async with ClientSession(read, write) as session:
            await session.initialize()
            
            # List available tools
            tools = await session.list_tools()
            print(f"Available tools: {[tool.name for tool in tools.tools]}")
            
            # Create a task
            result = await session.call_tool("create_task", {
                "title": "AI-generated task",
                "description": "Task created by AI client via HTTP",
                "priority": "medium"
            })
            print(f"Created task: {result}")

asyncio.run(main())

Hierarchical Task Management

# Create parent task via MCP
parent_result = await mcp_client.call_tool("create_task", {
    "title": "Implement User Management System",
    "description": "Complete user management functionality",
    "priority": "high",
    "tags": ["epic", "user-management"]
})

parent_id = parent_result["task"]["id"]

# Create child tasks
child_tasks = [
    {
        "title": "Design User Database Schema",
        "description": "Create tables for users, roles, permissions",
        "priority": "high",
        "parent_id": parent_id,
        "tags": ["database", "design"]
    },
    {
        "title": "Implement User Authentication",
        "description": "Add login/logout functionality",
        "priority": "medium", 
        "parent_id": parent_id,
        "tags": ["auth", "security"]
    }
]

for task_data in child_tasks:
    await mcp_client.call_tool("create_task", task_data)

HTTP REST API Usage

import httpx

# Create task via REST API
async with httpx.AsyncClient() as client:
    response = await client.post("http://localhost:8000/tools/create_task", json={
        "title": "REST API Task",
        "description": "Task created via HTTP API",
        "priority": "medium",
        "tags": ["api", "http"]
    })
    
    result = response.json()
    print(f"Task created: {result['result']}")

Server-Sent Events (SSE) Usage

// Connect to SSE endpoint for real-time updates
const eventSource = new EventSource('http://localhost:8000/events');

// Listen for task creation events
eventSource.addEventListener('task_created', function(event) {
    const data = JSON.parse(event.data);
    console.log('New task created:', data.task.title);
});

// Listen for task updates
eventSource.addEventListener('task_updated', function(event) {
    const data = JSON.parse(event.data);
    console.log('Task updated:', data.task.title);
});

// Listen for tool executions
eventSource.addEventListener('tool_executed', function(event) {
    const data = JSON.parse(event.data);
    console.log('Tool executed:', data.tool);
});

WebSocket Usage

// Connect to WebSocket for bidirectional communication
const ws = new WebSocket('ws://localhost:8000/ws');

ws.onopen = function() {
    console.log('WebSocket connected');
    
    // Execute tool via WebSocket
    ws.send(JSON.stringify({
        type: 'tool_call',
        tool: 'create_task',
        parameters: {
            title: 'WebSocket Task',
            description: 'Task created via WebSocket'
        }
    }));
};

ws.onmessage = function(event) {
    const data = JSON.parse(event.data);
    
    if (data.type === 'tool_result') {
        console.log('Tool result:', data.result);
    } else if (data.type === 'tool_executed') {
        console.log('Tool executed by another client:', data.tool);
    }
};

Streaming Operations

import httpx

# Stream task creation with progress updates
async with httpx.AsyncClient() as client:
    async with client.stream("POST", "http://localhost:8000/stream", json={
        "tool": "create_task",
        "parameters": {
            "title": "Streaming Task",
            "description": "Task with streaming progress"
        }
    }) as response:
        async for chunk in response.aiter_text():
            if chunk.startswith('data: '):
                data = json.loads(chunk[6:])
                
                if data.get('status') == 'processing':
                    print(f"Progress: {data.get('progress', 0)}%")
                elif data.get('status') == 'completed':
                    print(f"Task created: {data.get('result')}")

Advanced Task Filtering

# Search for high-priority authentication tasks
search_result = await mcp_client.call_tool("filter_tasks", {
    "priority": ["high", "urgent"],
    "tags": ["authentication"],
    "status": ["pending", "in_progress"],
    "limit": 50
})

# Get task statistics
stats = await mcp_client.call_tool("get_task_statistics", {})
print(f"Total tasks: {stats['statistics']['total_tasks']}")
print(f"Completion rate: {stats['statistics']['completion_rate']:.1%}")

Bulk Operations

# Update multiple tasks to completed status
task_ids = ["task-001", "task-002", "task-003"]
await mcp_client.call_tool("bulk_status_update", {
    "task_ids": task_ids,
    "status": "completed"
})

# Get complete hierarchy for reporting
hierarchy = await mcp_client.call_tool("get_task_hierarchy", {
    "root_id": parent_id
})

🏗️ Architecture

System Components

┌─────────────────┐    ┌─────────────────┐    ┌─────────────────┐
│   AI Agents     │    │  MCP Python SDK │    │  Todo Server    │
│                 │◄──►│                 │◄──►│                 │
│ - Claude        │    │ - Official SDK  │    │ - Low-level API │
│ - GPT-4         │    │ - stdio_server  │    │ - Lifespan Mgmt │
│ - Custom Agents │    │ - Tool Registry │    │ - HTTP Bridge   │
└─────────────────┘    └─────────────────┘    └─────────────────┘
                                                        │
                                                        ▼
                                               ┌─────────────────┐
                                               │  Storage Layer  │
                                               │                 │
                                               │ - Markdown Files│
                                               │ - YAML Metadata │
                                               │ - Real-time SSE │
                                               └─────────────────┘

Data Flow

1. AI Agent Request → MCP Protocol validation
2. Tool Execution → Business logic processing
3. Data Persistence → Markdown file storage
4. Response → Structured JSON response
5. Caching → Performance optimization
6. Audit Trail → Tool call logging

🚀 Deployment

Docker Deployment

FROM python:3.11-slim

WORKDIR /app
COPY . .

RUN pip install uv
RUN uv sync --frozen

EXPOSE 8000
CMD ["uv", "run", "todo-mcp-server", "--host", "0.0.0.0"]

Production Configuration

# Environment variables for production
export TODO_MCP_ENVIRONMENT=production
export TODO_MCP_DATA_DIRECTORY=/data/tasks
export TODO_MCP_LOG_LEVEL=INFO
export TODO_MCP_LOG_FILE=/var/log/todo-mcp.log
export TODO_MCP_BACKUP_ENABLED=true
export TODO_MCP_PERFORMANCE_MONITORING=true
export TODO_MCP_MAX_CACHE_SIZE=5000

# Start server
uv run todo-mcp-server

🔧 Troubleshooting

Common Issues

Performance Issues

# Check task statistics
uv run todo-mcp-server config-info

# Run performance tests
uv run pytest tests/test_integration/test_performance_load.py -v

# Enable performance monitoring
export TODO_MCP_PERFORMANCE_MONITORING=true

File System Issues

# Validate configuration
uv run todo-mcp-server validate

# Check file permissions
ls -la data/tasks/

# Repair corrupted files
uv run python -c "
from src.todo_mcp.storage.file_manager import FileManager
fm = FileManager('data')
fm.repair_corrupted_files()
"

Memory Issues

# Monitor memory usage
uv run pytest tests/test_integration/test_performance_load.py::TestMemoryUsage -v

# Adjust cache size
export TODO_MCP_MAX_CACHE_SIZE=500

📊 Monitoring & Metrics

Built-in Metrics

• Response time tracking
• Memory usage monitoring
• Cache hit/miss ratios
• Error rate tracking
• Concurrent user metrics

Health Checks

# Basic health check
curl -X POST http://localhost:8000/health

# Detailed system status
uv run todo-mcp-server config-info

🤝 Contributing

Development Setup

1. Fork the repository
2. Create a feature branch: git checkout -b feature/amazing-feature
3. Install development dependencies: uv sync --dev
4. Make your changes
5. Add tests for new functionality
6. Run the test suite: uv run pytest
7. Ensure code quality: uv run black . && uv run ruff check .
8. Submit a pull request

Code Standards

• Follow PEP 8 style guidelines
• Use type hints for all functions
• Write comprehensive tests (aim for >90% coverage)
• Document all public APIs
• Use Pydantic models for data validation

Testing Requirements

• Unit tests for all new functionality
• Integration tests for MCP tool interactions
• Performance tests for optimization changes
• Documentation updates for user-facing changes

📄 License

MIT License - see LICENSE file for details.

🆘 Support

• Issues: Use the GitHub issue tracker
• Discussions: Join our GitHub Discussions
• Documentation: Check the Wiki
• Security: Report security issues to security@your-domain.com

🙏 Acknowledgments

• Model Context Protocol for the specification
• Pydantic for data validation
• uv for fast Python package management
• The AI development community for feedback and contributions