MCP Personal Assistant Agent

from typing import List, Dict, Any, Optional
import os
import json
import logging
import re
from datetime import datetime
import httpx

# Import the MCP server instance from the main file
from mcp_server import mcp, Context

logger = logging.getLogger("mcp-pa-agent.tasks")

# Mock tasks database (in a real implementation, this would be a proper database)
TASKS_FILE = "tasks_data.json"

async def get_tasks():
    """Get all tasks from the storage."""
    try:
        if os.path.exists(TASKS_FILE):
            with open(TASKS_FILE, 'r') as f:
                return json.load(f)
        return []
    except Exception as e:
        logger.error(f"Error loading tasks: {str(e)}")
        return []

async def save_tasks(tasks):
    """Save tasks to the storage."""
    try:
        with open(TASKS_FILE, 'w') as f:
            json.dump(tasks, f, indent=2)
        return True
    except Exception as e:
        logger.error(f"Error saving tasks: {str(e)}")
        return False

# Resources
@mcp.resource("tasks://all")
async def all_tasks_resource() -> str:
    """Provide all tasks as a resource"""
    tasks = await get_tasks()
    return json.dumps(tasks, indent=2)

@mcp.resource("tasks://status/{status}")
async def tasks_by_status_resource(status: str) -> str:
    """Provide tasks filtered by status"""
    all_tasks = await get_tasks()
    filtered = [t for t in all_tasks if t.get('status', '').lower() == status.lower()]
    return json.dumps(filtered, indent=2)

@mcp.resource("tasks://priority/{priority}")
async def tasks_by_priority_resource(priority: str) -> str:
    """Provide tasks filtered by priority"""
    all_tasks = await get_tasks()
    filtered = [t for t in all_tasks if t.get('priority', '').lower() == priority.lower()]
    return json.dumps(filtered, indent=2)

# Prompts
@mcp.prompt()
def create_task_prompt(description: str, priority: str = "medium") -> str:
    """Create a prompt for task creation"""
    return f"Please help me create a new task based on this description: '{description}'. It should have {priority} priority. Please organize the information into title, description, and suggest a reasonable due date if applicable."

# Tool functions
@mcp.tool()
async def list_tasks(status: str = "all", ctx: Context = None) -> str:
    """List tasks with optional filtering by status.
    
    Args:
        status: Filter tasks by status ('all', 'pending', 'completed', 'in_progress')
    """
    if ctx:
        ctx.info(f"Listing tasks with status filter: {status}")
    
    tasks = await get_tasks()
    
    if not tasks:
        return "No tasks found."
    
    filtered_tasks = tasks
    if status.lower() != "all":
        filtered_tasks = [task for task in tasks if task.get('status', '').lower() == status.lower()]
    
    if not filtered_tasks:
        return f"No tasks with status '{status}' found."
    
    formatted_tasks = []
    for idx, task in enumerate(filtered_tasks, 1):
        due_date = task.get('due_date', 'No due date')
        created_at = task.get('created_at', 'Unknown')
        updated_at = task.get('updated_at', created_at)
        
        formatted_tasks.append(f"""
Task #{task.get('id', idx)}: {task.get('title', 'Untitled')}
Status: {task.get('status', 'Not specified')}
Priority: {task.get('priority', 'Not specified')}
Due Date: {due_date}
Description: {task.get('description', 'No description provided')}
Created: {created_at}
Last Updated: {updated_at}
""")
    
    return "\n---\n".join(formatted_tasks)

@mcp.tool()
async def add_task(title: str, description: str = "", priority: str = "medium", due_date: str = "", ctx: Context = None) -> str:
    """Add a new task.
    
    Args:
        title: Title of the task
        description: Optional description of the task
        priority: Priority level ('low', 'medium', 'high')
        due_date: Optional due date in YYYY-MM-DD format
    """
    if ctx:
        ctx.info(f"Adding new task: {title}")
    
    tasks = await get_tasks()
    
    # Validate title
    if not title or len(title.strip()) == 0:
        error_msg = "Error: Task title cannot be empty."
        if ctx:
            ctx.error(error_msg)
        return error_msg
    
    # Validate priority
    valid_priorities = ['low', 'medium', 'high']
    if priority.lower() not in valid_priorities:
        error_msg = f"Invalid priority: {priority}. Must be one of: {', '.join(valid_priorities)}."
        if ctx:
            ctx.error(error_msg)
        return error_msg
    
    # Validate due date format if provided
    if due_date:
        date_pattern = re.compile(r'^\d{4}-\d{2}-\d{2}$')
        if not date_pattern.match(due_date):
            error_msg = f"Invalid due date format: {due_date}. Please use YYYY-MM-DD."
            if ctx:
                ctx.error(error_msg)
            return error_msg
            
        try:
            datetime.strptime(due_date, '%Y-%m-%d')
        except ValueError:
            error_msg = f"Invalid due date: {due_date}. Please use a valid date in YYYY-MM-DD format."
            if ctx:
                ctx.error(error_msg)
            return error_msg
    
    # Generate a unique ID
    task_id = 1
    if tasks:
        existing_ids = [task.get('id', 0) for task in tasks]
        task_id = max(existing_ids) + 1
    
    new_task = {
        'id': task_id,
        'title': title,
        'description': description,
        'status': 'pending',
        'priority': priority.lower(),
        'due_date': due_date,
        'created_at': datetime.now().isoformat(),
    }
    
    tasks.append(new_task)
    if await save_tasks(tasks):
        success_msg = f"Task '{title}' added successfully with ID {new_task['id']}."
        if ctx:
            ctx.info(success_msg)
        return success_msg
    else:
        error_msg = "Failed to save the task. Please try again."
        if ctx:
            ctx.error(error_msg)
        return error_msg

@mcp.tool()
async def update_task_status(task_id: int, status: str, ctx: Context = None) -> str:
    """Update the status of a task.
    
    Args:
        task_id: The ID of the task to update
        status: New status ('pending', 'in_progress', 'completed')
    """
    if ctx:
        ctx.info(f"Updating task #{task_id} status to: {status}")
    
    # Validate status
    valid_statuses = ['pending', 'in_progress', 'completed']
    if status.lower() not in valid_statuses:
        error_msg = f"Invalid status: {status}. Must be one of: {', '.join(valid_statuses)}."
        if ctx:
            ctx.error(error_msg)
        return error_msg
    
    tasks = await get_tasks()
    
    # Find the task with the given ID
    task_found = False
    for task in tasks:
        if task.get('id') == task_id:
            task_found = True
            old_status = task.get('status', 'unknown')
            task['status'] = status.lower()
            task['updated_at'] = datetime.now().isoformat()
            
            if await save_tasks(tasks):
                success_msg = f"Task #{task_id} status updated from '{old_status}' to '{status}'."
                if ctx:
                    ctx.info(success_msg)
                return success_msg
            else:
                error_msg = "Failed to update task status. Please try again."
                if ctx:
                    ctx.error(error_msg)
                return error_msg
    
    error_msg = f"Task with ID {task_id} not found."
    if ctx:
        ctx.error(error_msg)
    return error_msg

@mcp.tool()
async def delete_task(task_id: int, ctx: Context = None) -> str:
    """Delete a task.
    
    Args:
        task_id: The ID of the task to delete
    """
    if ctx:
        ctx.info(f"Deleting task #{task_id}")
    
    tasks = await get_tasks()
    
    # Find and remove the task with the given ID
    original_count = len(tasks)
    tasks = [task for task in tasks if task.get('id') != task_id]
    
    if len(tasks) < original_count:
        if await save_tasks(tasks):
            success_msg = f"Task #{task_id} deleted successfully."
            if ctx:
                ctx.info(success_msg)
            return success_msg
        else:
            error_msg = "Failed to delete task. Please try again."
            if ctx:
                ctx.error(error_msg)
            return error_msg
    else:
        error_msg = f"Task with ID {task_id} not found."
        if ctx:
            ctx.error(error_msg)
        return error_msg