Add MCP task manager server with Gradio interface

app.py

@@ -0,0 +1,61 @@
+import gradio as gr
+import json
+from task_manager import TaskManagerServer
+import asyncio
+
+# Create MCP server instance
+server = TaskManagerServer()
+
+async def call_mcp_tool(tool_name, arguments_json):
+    """Call MCP tools via web interface"""
+    try:
+        args = json.loads(arguments_json) if arguments_json else {}
+        
+        if tool_name == "add_task":
+            result = await server.add_task(args.get("title", ""), args.get("description", ""))
+        elif tool_name == "list_tasks":
+            result = await server.list_tasks()
+        elif tool_name == "complete_task":
+            result = await server.complete_task(args.get("task_id", 0))
+        elif tool_name == "delete_task":
+            result = await server.delete_task(args.get("task_id", 0))
+        else:
+            return "Unknown tool"
+            
+        return json.dumps(result, indent=2)
+    except Exception as e:
+        return f"Error: {str(e)}"
+
+def sync_call_mcp_tool(tool_name, arguments_json):
+    """Sync wrapper for async function"""
+    return asyncio.run(call_mcp_tool(tool_name, arguments_json))
+
+# Create Gradio interface
+with gr.Blocks(title="Task Manager MCP Server") as demo:
+    gr.Markdown("# Task Manager MCP Server")
+    gr.Markdown("Test your MCP server tools through this web interface")
+    
+    with gr.Row():
+        tool_dropdown = gr.Dropdown(
+            choices=["add_task", "list_tasks", "complete_task", "delete_task"],
+            label="Select Tool",
+            value="list_tasks"
+        )
+        
+    arguments_input = gr.Textbox(
+        label="Arguments (JSON)",
+        placeholder='{"title": "My Task", "description": "Task description"}',
+        lines=3
+    )
+    
+    submit_btn = gr.Button("Call Tool")
+    output = gr.Textbox(label="Result", lines=10)
+    
+    submit_btn.click(
+        sync_call_mcp_tool,
+        inputs=[tool_dropdown, arguments_input],
+        outputs=output
+    )
+
+if __name__ == "__main__":
+    demo.launch()

requirements.txt

@@ -0,0 +1,3 @@
+gradio
+mcp
+pydantic

task_manager.py

@@ -0,0 +1,501 @@
+"""
+Task Manager MCP Server - A comprehensive example for learning MCP development
+
+This server provides task management functionality with multiple tools:
+- Create, read, update, delete tasks
+- List tasks with filtering
+- Mark tasks as complete/incomplete
+- Add tags and priorities
+- Search functionality
+
+Perfect for learning MCP concepts and extending with new features!
+"""
+
+import json
+import os
+from datetime import datetime
+from typing import Any, Dict, List, Optional
+from enum import Enum
+
+from mcp.server.fastmcp import FastMCP
+from pydantic import BaseModel
+
+# Initialize FastMCP server
+mcp = FastMCP("task-manager")
+
+# Data models
+class Priority(str, Enum):
+    LOW = "low"
+    MEDIUM = "medium"
+    HIGH = "high"
+    URGENT = "urgent"
+
+class TaskStatus(str, Enum):
+    PENDING = "pending"
+    IN_PROGRESS = "in_progress"
+    COMPLETED = "completed"
+    CANCELLED = "cancelled"
+
+class Task(BaseModel):
+    id: int
+    title: str
+    description: str = ""
+    status: TaskStatus = TaskStatus.PENDING
+    priority: Priority = Priority.MEDIUM
+    tags: List[str] = []
+    created_at: str
+    updated_at: str
+    due_date: Optional[str] = None
+    completed_at: Optional[str] = None
+
+# In-memory storage (in a real app, you'd use a database)
+tasks_db: Dict[int, Task] = {}
+next_task_id = 1
+
+# File-based persistence
+DATA_FILE = "tasks.json"
+
+def load_tasks():
+    """Load tasks from file if it exists"""
+    global tasks_db, next_task_id
+    
+    if os.path.exists(DATA_FILE):
+        try:
+            with open(DATA_FILE, 'r') as f:
+                data = json.load(f)
+                tasks_db = {int(k): Task(**v) for k, v in data.get('tasks', {}).items()}
+                next_task_id = data.get('next_id', 1)
+        except Exception as e:
+            print(f"Error loading tasks: {e}")
+
+def save_tasks():
+    """Save tasks to file"""
+    try:
+        data = {
+            'tasks': {str(k): v.model_dump() for k, v in tasks_db.items()},
+            'next_id': next_task_id
+        }
+        with open(DATA_FILE, 'w') as f:
+            json.dump(data, f, indent=2)
+    except Exception as e:
+        print(f"Error saving tasks: {e}")
+
+def get_current_time() -> str:
+    """Get current timestamp as ISO string"""
+    return datetime.now().isoformat()
+
+# Load existing tasks on startup
+load_tasks()
+
+@mcp.tool()
+def create_task(
+    title: str,
+    description: str = "",
+    priority: str = "medium",
+    tags: str = "",
+    due_date: str = ""
+) -> str:
+    """Create a new task.
+    
+    Args:
+        title: Task title (required)
+        description: Detailed description of the task
+        priority: Task priority (low, medium, high, urgent)
+        tags: Comma-separated list of tags
+        due_date: Due date in YYYY-MM-DD format
+    """
+    global next_task_id
+    
+    try:
+        # Parse priority
+        task_priority = Priority(priority.lower())
+    except ValueError:
+        return f"Invalid priority '{priority}'. Use: low, medium, high, urgent"
+    
+    # Parse tags
+    task_tags = [tag.strip() for tag in tags.split(",") if tag.strip()]
+    
+    # Validate due date if provided
+    task_due_date = None
+    if due_date:
+        try:
+            datetime.fromisoformat(due_date)
+            task_due_date = due_date
+        except ValueError:
+            return f"Invalid due date format. Use YYYY-MM-DD"
+    
+    # Create task
+    current_time = get_current_time()
+    task = Task(
+        id=next_task_id,
+        title=title,
+        description=description,
+        priority=task_priority,
+        tags=task_tags,
+        created_at=current_time,
+        updated_at=current_time,
+        due_date=task_due_date
+    )
+    
+    tasks_db[next_task_id] = task
+    next_task_id += 1
+    save_tasks()
+    
+    return f"✅ Created task #{task.id}: {task.title}"
+
+@mcp.tool()
+def list_tasks(
+    status: str = "all",
+    priority: str = "all",
+    tag: str = "",
+    limit: int = 10
+) -> str:
+    """List tasks with optional filtering.
+    
+    Args:
+        status: Filter by status (all, pending, in_progress, completed, cancelled)
+        priority: Filter by priority (all, low, medium, high, urgent)
+        tag: Filter by tag (exact match)
+        limit: Maximum number of tasks to return
+    """
+    if not tasks_db:
+        return "📝 No tasks found. Create your first task!"
+    
+    filtered_tasks = list(tasks_db.values())
+    
+    # Filter by status
+    if status != "all":
+        try:
+            status_filter = TaskStatus(status)
+            filtered_tasks = [t for t in filtered_tasks if t.status == status_filter]
+        except ValueError:
+            return f"Invalid status '{status}'. Use: all, pending, in_progress, completed, cancelled"
+    
+    # Filter by priority
+    if priority != "all":
+        try:
+            priority_filter = Priority(priority)
+            filtered_tasks = [t for t in filtered_tasks if t.priority == priority_filter]
+        except ValueError:
+            return f"Invalid priority '{priority}'. Use: all, low, medium, high, urgent"
+    
+    # Filter by tag
+    if tag:
+        filtered_tasks = [t for t in filtered_tasks if tag in t.tags]
+    
+    # Sort by creation date (newest first)
+    filtered_tasks.sort(key=lambda x: x.created_at, reverse=True)
+    
+    # Apply limit
+    filtered_tasks = filtered_tasks[:limit]
+    
+    if not filtered_tasks:
+        return "📝 No tasks match your filters."
+    
+    # Format output
+    result = f"📋 Found {len(filtered_tasks)} task(s):\n\n"
+    
+    for task in filtered_tasks:
+        status_emoji = {
+            TaskStatus.PENDING: "⏳",
+            TaskStatus.IN_PROGRESS: "🔄",
+            TaskStatus.COMPLETED: "✅",
+            TaskStatus.CANCELLED: "❌"
+        }
+        
+        priority_emoji = {
+            Priority.LOW: "🟢",
+            Priority.MEDIUM: "🟡",
+            Priority.HIGH: "🟠",
+            Priority.URGENT: "🔴"
+        }
+        
+        tags_str = f" #{' #'.join(task.tags)}" if task.tags else ""
+        due_str = f" (Due: {task.due_date})" if task.due_date else ""
+        
+        result += f"{status_emoji[task.status]} #{task.id} - {task.title}\n"
+        result += f"   {priority_emoji[task.priority]} {task.priority.value.title()}{due_str}{tags_str}\n"
+        if task.description:
+            result += f"   📄 {task.description}\n"
+        result += "\n"
+    
+    return result.strip()
+
+@mcp.tool()
+def get_task(task_id: int) -> str:
+    """Get detailed information about a specific task.
+    
+    Args:
+        task_id: The ID of the task to retrieve
+    """
+    if task_id not in tasks_db:
+        return f"❌ Task #{task_id} not found."
+    
+    task = tasks_db[task_id]
+    
+    status_emoji = {
+        TaskStatus.PENDING: "⏳",
+        TaskStatus.IN_PROGRESS: "🔄",
+        TaskStatus.COMPLETED: "✅",
+        TaskStatus.CANCELLED: "❌"
+    }
+    
+    priority_emoji = {
+        Priority.LOW: "🟢",
+        Priority.MEDIUM: "🟡",
+        Priority.HIGH: "🟠",
+        Priority.URGENT: "🔴"
+    }
+    
+    result = f"📋 Task #{task.id}\n"
+    result += f"Title: {task.title}\n"
+    result += f"Status: {status_emoji[task.status]} {task.status.value.replace('_', ' ').title()}\n"
+    result += f"Priority: {priority_emoji[task.priority]} {task.priority.value.title()}\n"
+    
+    if task.description:
+        result += f"Description: {task.description}\n"
+    
+    if task.tags:
+        result += f"Tags: #{' #'.join(task.tags)}\n"
+    
+    if task.due_date:
+        result += f"Due Date: {task.due_date}\n"
+    
+    result += f"Created: {task.created_at}\n"
+    result += f"Updated: {task.updated_at}\n"
+    
+    if task.completed_at:
+        result += f"Completed: {task.completed_at}\n"
+    
+    return result
+
+@mcp.tool()
+def update_task(
+    task_id: int,
+    title: str = "",
+    description: str = "",
+    status: str = "",
+    priority: str = "",
+    tags: str = "",
+    due_date: str = ""
+) -> str:
+    """Update an existing task. Only provided fields will be updated.
+    
+    Args:
+        task_id: The ID of the task to update
+        title: New task title
+        description: New task description
+        status: New status (pending, in_progress, completed, cancelled)
+        priority: New priority (low, medium, high, urgent)
+        tags: New comma-separated list of tags
+        due_date: New due date in YYYY-MM-DD format (use 'none' to clear)
+    """
+    if task_id not in tasks_db:
+        return f"❌ Task #{task_id} not found."
+    
+    task = tasks_db[task_id]
+    updated_fields = []
+    
+    # Update title
+    if title:
+        task.title = title
+        updated_fields.append("title")
+    
+    # Update description
+    if description:
+        task.description = description
+        updated_fields.append("description")
+    
+    # Update status
+    if status:
+        try:
+            new_status = TaskStatus(status)
+            task.status = new_status
+            updated_fields.append("status")
+            
+            # Set completion time if marking as completed
+            if new_status == TaskStatus.COMPLETED and not task.completed_at:
+                task.completed_at = get_current_time()
+            elif new_status != TaskStatus.COMPLETED:
+                task.completed_at = None
+                
+        except ValueError:
+            return f"Invalid status '{status}'. Use: pending, in_progress, completed, cancelled"
+    
+    # Update priority
+    if priority:
+        try:
+            task.priority = Priority(priority)
+            updated_fields.append("priority")
+        except ValueError:
+            return f"Invalid priority '{priority}'. Use: low, medium, high, urgent"
+    
+    # Update tags
+    if tags:
+        if tags.lower() == "none":
+            task.tags = []
+        else:
+            task.tags = [tag.strip() for tag in tags.split(",") if tag.strip()]
+        updated_fields.append("tags")
+    
+    # Update due date
+    if due_date:
+        if due_date.lower() == "none":
+            task.due_date = None
+        else:
+            try:
+                datetime.fromisoformat(due_date)
+                task.due_date = due_date
+            except ValueError:
+                return f"Invalid due date format. Use YYYY-MM-DD or 'none' to clear"
+        updated_fields.append("due_date")
+    
+    if not updated_fields:
+        return f"❌ No fields provided to update for task #{task_id}."
+    
+    # Update timestamp
+    task.updated_at = get_current_time()
+    save_tasks()
+    
+    return f"✅ Updated task #{task_id}: {', '.join(updated_fields)}"
+
+@mcp.tool()
+def delete_task(task_id: int) -> str:
+    """Delete a task permanently.
+    
+    Args:
+        task_id: The ID of the task to delete
+    """
+    if task_id not in tasks_db:
+        return f"❌ Task #{task_id} not found."
+    
+    task = tasks_db[task_id]
+    del tasks_db[task_id]
+    save_tasks()
+    
+    return f"🗑️ Deleted task #{task_id}: {task.title}"
+
+@mcp.tool()
+def search_tasks(query: str, limit: int = 10) -> str:
+    """Search tasks by title, description, or tags.
+    
+    Args:
+        query: Search query (searches in title, description, and tags)
+        limit: Maximum number of results to return
+    """
+    if not tasks_db:
+        return "📝 No tasks to search."
+    
+    query_lower = query.lower()
+    matching_tasks = []
+    
+    for task in tasks_db.values():
+        # Search in title, description, and tags
+        if (query_lower in task.title.lower() or 
+            query_lower in task.description.lower() or
+            any(query_lower in tag.lower() for tag in task.tags)):
+            matching_tasks.append(task)
+    
+    if not matching_tasks:
+        return f"🔍 No tasks found matching '{query}'."
+    
+    # Sort by relevance (title matches first, then description, then tags)
+    def relevance_score(task):
+        score = 0
+        if query_lower in task.title.lower():
+            score += 3
+        if query_lower in task.description.lower():
+            score += 2
+        if any(query_lower in tag.lower() for tag in task.tags):
+            score += 1
+        return score
+    
+    matching_tasks.sort(key=relevance_score, reverse=True)
+    matching_tasks = matching_tasks[:limit]
+    
+    result = f"🔍 Found {len(matching_tasks)} task(s) matching '{query}':\n\n"
+    
+    for task in matching_tasks:
+        status_emoji = {
+            TaskStatus.PENDING: "⏳",
+            TaskStatus.IN_PROGRESS: "🔄",
+            TaskStatus.COMPLETED: "✅",
+            TaskStatus.CANCELLED: "❌"
+        }
+        
+        result += f"{status_emoji[task.status]} #{task.id} - {task.title}\n"
+        if task.description:
+            result += f"   📄 {task.description[:100]}{'...' if len(task.description) > 100 else ''}\n"
+        result += "\n"
+    
+    return result.strip()
+
+@mcp.tool()
+def get_stats() -> str:
+    """Get task statistics and summary."""
+    if not tasks_db:
+        return "📊 No tasks yet. Create your first task to see statistics!"
+    
+    total_tasks = len(tasks_db)
+    
+    # Count by status
+    status_counts = {}
+    for status in TaskStatus:
+        status_counts[status] = sum(1 for task in tasks_db.values() if task.status == status)
+    
+    # Count by priority
+    priority_counts = {}
+    for priority in Priority:
+        priority_counts[priority] = sum(1 for task in tasks_db.values() if task.priority == priority)
+    
+    # Overdue tasks
+    today = datetime.now().date()
+    overdue_count = 0
+    for task in tasks_db.values():
+        if (task.due_date and task.status != TaskStatus.COMPLETED and 
+            datetime.fromisoformat(task.due_date).date() < today):
+            overdue_count += 1
+    
+    # Most common tags
+    all_tags = []
+    for task in tasks_db.values():
+        all_tags.extend(task.tags)
+    
+    tag_counts = {}
+    for tag in all_tags:
+        tag_counts[tag] = tag_counts.get(tag, 0) + 1
+    
+    top_tags = sorted(tag_counts.items(), key=lambda x: x[1], reverse=True)[:5]
+    
+    result = f"📊 Task Statistics\n\n"
+    result += f"Total Tasks: {total_tasks}\n\n"
+    
+    result += "📈 By Status:\n"
+    for status, count in status_counts.items():
+        emoji = {"pending": "⏳", "in_progress": "🔄", "completed": "✅", "cancelled": "❌"}
+        result += f"  {emoji.get(status.value, '•')} {status.value.replace('_', ' ').title()}: {count}\n"
+    
+    result += "\n🎯 By Priority:\n"
+    for priority, count in priority_counts.items():
+        emoji = {"low": "🟢", "medium": "🟡", "high": "🟠", "urgent": "🔴"}
+        result += f"  {emoji.get(priority.value, '•')} {priority.value.title()}: {count}\n"
+    
+    if overdue_count > 0:
+        result += f"\n⚠️ Overdue Tasks: {overdue_count}\n"
+    
+    if top_tags:
+        result += f"\n🏷️ Top Tags:\n"
+        for tag, count in top_tags:
+            result += f"  #{tag}: {count}\n"
+    
+    return result
+
+def main():
+    """Initialize and run the MCP server"""
+    print("Starting Task Manager MCP Server...")
+    print(f"Data file: {os.path.abspath(DATA_FILE)}")
+    print(f"Loaded {len(tasks_db)} existing tasks")
+    mcp.run(transport="stdio")
+
+if __name__ == "__main__":
+    main()

tasks.json

@@ -0,0 +1,65 @@
+{
+  "tasks": {
+    "1": {
+      "id": 1,
+      "title": "Learn MCP development",
+      "description": "Study Model Context Protocol",
+      "status": "in_progress",
+      "priority": "high",
+      "tags": [
+        "learning",
+        "mcp"
+      ],
+      "created_at": "2025-12-16T01:47:25.373667",
+      "updated_at": "2025-12-16T01:47:25.375669",
+      "due_date": null,
+      "completed_at": null
+    },
+    "2": {
+      "id": 2,
+      "title": "Build a web app",
+      "description": "Create a task management web interface",
+      "status": "completed",
+      "priority": "medium",
+      "tags": [
+        "web",
+        "frontend"
+      ],
+      "created_at": "2025-12-16T01:47:25.374668",
+      "updated_at": "2025-12-16T01:47:25.376669",
+      "due_date": "2024-12-25",
+      "completed_at": "2025-12-16T01:47:25.376669"
+    },
+    "3": {
+      "id": 3,
+      "title": "Write documentation",
+      "description": "Document the MCP server",
+      "status": "pending",
+      "priority": "low",
+      "tags": [
+        "docs"
+      ],
+      "created_at": "2025-12-16T01:47:25.374668",
+      "updated_at": "2025-12-16T01:47:25.374668",
+      "due_date": null,
+      "completed_at": null
+    },
+    "4": {
+      "id": 4,
+      "title": "Learn React",
+      "description": "Study React fundamentals, components, hooks, and state management",
+      "status": "pending",
+      "priority": "high",
+      "tags": [
+        "frontend",
+        "react",
+        "learning"
+      ],
+      "created_at": "2025-12-16T01:53:44.607322",
+      "updated_at": "2025-12-16T01:53:44.607322",
+      "due_date": null,
+      "completed_at": null
+    }
+  },
+  "next_id": 5
+}