llm-todo-mcp-mao 0.1.1
A todo application MCP server designed for AI agent task management
License: MIT
Requires Python: >=3.10

# 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:
```bash
git clone <repository-url>
cd todo-mcp
```

2. Create virtual environment and install dependencies:
```bash
uv sync
```

3. Install development dependencies:
```bash
uv sync --dev
```

## Usage

### Running the Servers

#### MCP Server (stdio protocol)
Start the MCP server with default settings:
```bash
uv run todo-mcp-server serve
```

With custom configuration:
```bash
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:
```bash
uv run todo-mcp-server serve-fastmcp
```

With custom host and port:
```bash
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:
```bash
uv run todo-mcp-server serve-http
```

With custom host and port:
```bash
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:
```bash
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
```bash
uv run pytest
```

#### Test with Coverage
```bash
uv run pytest --cov=src/todo_mcp --cov-report=html
```

#### Test Categories
```bash
# 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
```bash
# 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:
```bash
uv run black .
uv run isort .
```

Lint code:
```bash
uv run ruff check .
```

Type checking:
```bash
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:

```env
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

```bash
# 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:

```markdown
---
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)
```python
# 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)
```python
# 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
```python
# 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

```python
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

```javascript
// 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

```javascript
// 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

```python
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

```python
# 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

```python
# 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

```dockerfile
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

```bash
# 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
```bash
# 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
```bash
# 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
```bash
# 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
```bash
# 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](LICENSE) file for details.

## 🆘 Support

- **Issues**: Use the [GitHub issue tracker](https://github.com/your-repo/todo-mcp/issues)
- **Discussions**: Join our [GitHub Discussions](https://github.com/your-repo/todo-mcp/discussions)
- **Documentation**: Check the [Wiki](https://github.com/your-repo/todo-mcp/wiki)
- **Security**: Report security issues to security@your-domain.com

## 🙏 Acknowledgments

- [Model Context Protocol](https://modelcontextprotocol.io/) for the specification
- [Pydantic](https://pydantic.dev/) for data validation
- [uv](https://github.com/astral-sh/uv) for fast Python package management
- The AI development community for feedback and contributions