小智AI - 日程管理MCP工具实现

admin

# 日程管理MCP工具实现

下面我将实现一个功能完整的日程管理MCP工具，支持日程记录、提醒、查询和删除等功能：

```python
# schedule_manager.py
from mcp.server.fastmcp import FastMCP
import sys
import logging
import json
import datetime
import time
import threading
from typing import List, Dict, Optional

logger = logging.getLogger('ScheduleManager')

# Fix UTF-8 encoding for Windows console
if sys.platform == 'win32':
    sys.stderr.reconfigure(encoding='utf-8')
    sys.stdout.reconfigure(encoding='utf-8')

# 创建MCP服务器实例
mcp = FastMCP("ScheduleManager")

# 内存存储日程数据（生产环境建议使用数据库）
schedules = []
schedule_lock = threading.Lock()
reminder_thread = None
stop_reminder_flag = False

class Schedule:
    """日程类，封装日程数据和逻辑"""
    def __init__(self, title: str, start_time: str, end_time: str = None,
                 reminder: str = None, description: str = ""):
        self.id = f"{int(time.time())}-{len(schedules)}"
        self.title = title
        self.description = description
        self.start_time = datetime.datetime.fromisoformat(start_time)
        
        if end_time:
            self.end_time = datetime.datetime.fromisoformat(end_time)
        else:
            self.end_time = None
            
        if reminder:
            self.reminder = datetime.datetime.fromisoformat(reminder)
        else:
            self.reminder = None
            
        self.reminded = False
   
    def to_dict(self) -> dict:
        """将日程对象转换为字典"""
        return {
            "id": self.id,
            "title": self.title,
            "description": self.description,
            "start_time": self.start_time.isoformat(),
            "end_time": self.end_time.isoformat() if self.end_time else None,
            "reminder": self.reminder.isoformat() if self.reminder else None,
            "reminded": self.reminded
        }
   
    def is_active(self) -> bool:
        """检查日程是否正在进行中"""
        now = datetime.datetime.now()
        return self.start_time <= now <= (self.end_time or self.start_time + datetime.timedelta(hours=1))
   
    def should_remind(self) -> bool:
        """检查是否需要发送提醒"""
        if not self.reminder or self.reminded:
            return False
            
        now = datetime.datetime.now()
        return now >= self.reminder and not self.reminded

# 添加日程工具
@mcp.tool()
def add_schedule(
    title: str,
    start_time: str,
    end_time: str = None,
    reminder: str = None,
    description: str = ""
) -> dict:
    """
    添加新的日程安排。日期时间格式应为ISO 8601格式（例如：2023-10-15T14:30:00）
   
    Args:
        title: 日程标题
        start_time: 开始时间（ISO格式）
        end_time: 结束时间（ISO格式，可选）
        reminder: 提醒时间（ISO格式，可选）
        description: 日程描述（可选）
   
    Returns:
        操作结果和日程ID
    """
    try:
        # 创建日程对象
        new_schedule = Schedule(title, start_time, end_time, reminder, description)
        
        # 线程安全地添加到日程列表
        with schedule_lock:
            schedules.append(new_schedule)
        
        logger.info(f"添加日程成功: {title} ({new_schedule.id})")
        return {"success": True, "message": "日程添加成功", "schedule_id": new_schedule.id}
   
    except Exception as e:
        logger.error(f"添加日程失败: {str(e)}")
        return {"success": False, "message": f"添加日程失败: {str(e)}"}

# 查询日程工具
@mcp.tool()
def query_schedules(
    date: str = None,
    keyword: str = None,
    upcoming: bool = False
) -> dict:
    """
    查询日程安排。可以按日期、关键词或查询即将到来的日程
   
    Args:
        date: 查询日期（格式：YYYY-MM-DD）
        keyword: 搜索关键词
        upcoming: 是否查询即将到来的日程（默认False）
   
    Returns:
        匹配的日程列表
    """
    try:
        result = []
        now = datetime.datetime.now()
        
        with schedule_lock:
            for schedule in schedules:
                # 按日期筛选
                if date:
                    schedule_date = schedule.start_time.date().isoformat()
                    if schedule_date != date:
                        continue
               
                # 按关键词筛选
                if keyword:
                    keyword = keyword.lower()
                    if (keyword not in schedule.title.lower() and
                        keyword not in schedule.description.lower()):
                        continue
               
                # 查询即将到来的日程（未来1周内）
                if upcoming:
                    if schedule.start_time < now or schedule.start_time > now + datetime.timedelta(days=7):
                        continue
               
                result.append(schedule.to_dict())
        
        logger.info(f"查询到 {len(result)} 条日程")
        return {"success": True, "count": len(result), "schedules": result}
   
    except Exception as e:
        logger.error(f"查询日程失败: {str(e)}")
        return {"success": False, "message": f"查询日程失败: {str(e)}"}

# 删除日程工具
@mcp.tool()
def delete_schedule(schedule_id: str) -> dict:
    """
    根据ID删除日程
   
    Args:
        schedule_id: 要删除的日程ID
   
    Returns:
        操作结果
    """
    try:
        global schedules
        
        with schedule_lock:
            # 查找并删除日程
            original_count = len(schedules)
            schedules = [s for s in schedules if s.id != schedule_id]
            deleted_count = original_count - len(schedules)
        
        if deleted_count > 0:
            logger.info(f"删除日程成功: {schedule_id}")
            return {"success": True, "message": "日程删除成功"}
        else:
            logger.warning(f"未找到日程: {schedule_id}")
            return {"success": False, "message": "未找到指定日程"}
   
    except Exception as e:
        logger.error(f"删除日程失败: {str(e)}")
        return {"success": False, "message": f"删除日程失败: {str(e)}"}

# 提醒功能工具
@mcp.tool()
def set_reminder(schedule_id: str, reminder_time: str) -> dict:
    """
    为日程设置或修改提醒时间
   
    Args:
        schedule_id: 日程ID
        reminder_time: 提醒时间（ISO格式）
   
    Returns:
        操作结果
    """
    try:
        with schedule_lock:
            for schedule in schedules:
                if schedule.id == schedule_id:
                    schedule.reminder = datetime.datetime.fromisoformat(reminder_time)
                    schedule.reminded = False
                    logger.info(f"为日程 {schedule_id} 设置提醒时间: {reminder_time}")
                    return {"success": True, "message": "提醒设置成功"}
        
        logger.warning(f"未找到日程: {schedule_id}")
        return {"success": False, "message": "未找到指定日程"}
   
    except Exception as e:
        logger.error(f"设置提醒失败: {str(e)}")
        return {"success": False, "message": f"设置提醒失败: {str(e)}"}

# 提醒后台线程
def reminder_daemon():
    """后台提醒线程，每分钟检查一次需要提醒的日程"""
    global stop_reminder_flag
   
    logger.info("日程提醒守护线程已启动")
   
    while not stop_reminder_flag:
        try:
            now = datetime.datetime.now()
            reminders_to_send = []
            
            # 收集需要发送的提醒
            with schedule_lock:
                for schedule in schedules:
                    if schedule.should_remind():
                        reminders_to_send.append(schedule)
                        schedule.reminded = True
            
            # 发送提醒
            for schedule in reminders_to_send:
                reminder_data = {
                    "event": "schedule_reminder",
                    "schedule_id": schedule.id,
                    "title": schedule.title,
                    "start_time": schedule.start_time.isoformat(),
                    "description": schedule.description
                }
                # 通过MCP发送提醒事件
                mcp.send_event("reminder", reminder_data)
                logger.warning(f"🔔 日程提醒: {schedule.title} ({schedule.start_time})")
        
        except Exception as e:
            logger.error(f"提醒线程异常: {str(e)}")
        
        # 每分钟检查一次
        time.sleep(60)
   
    logger.info("日程提醒守护线程已停止")

# 启动服务器
if __name__ == "__main__":
    # 启动提醒线程
    stop_reminder_flag = False
    reminder_thread = threading.Thread(target=reminder_daemon, daemon=True)
    reminder_thread.start()
   
    try:
        # 启动MCP服务器
        mcp.run(transport="stdio")
    finally:
        # 清理资源
        stop_reminder_flag = True
        if reminder_thread.is_alive():
            reminder_thread.join(timeout=5)
```

## 功能说明

### 核心功能
1. **日程记录**：
   - 添加包含标题、时间、描述和提醒的日程
   - 支持ISO 8601标准时间格式
   - 自动生成唯一ID标识每个日程

2. **智能提醒**：
   - 后台守护线程每分钟检查提醒
   - 使用MCP事件机制发送提醒通知
   - 避免重复提醒（reminded标志）

3. **日程查询**：
   - 按日期查询（YYYY-MM-DD）
   - 按关键词搜索
   - 查询即将到来的日程（未来一周内）

4. **日程管理**：
   - 根据ID删除日程
   - 修改提醒时间
   - 自动清理过期日程（可选扩展）

### 技术亮点

1. **线程安全设计**：
   - 使用`threading.Lock`确保多线程安全
   - 守护线程与主线程分离

2. **事件驱动架构**：
   ```python
   mcp.send_event("reminder", reminder_data)  # 发送提醒事件
   ```
   - 通过MCP事件机制实现主动通知

3. **时间智能处理**：
   - 自动转换ISO时间字符串
   - 计算日程活动状态
   - 智能提醒触发判断

4. **内存存储优化**：
   - 使用类封装日程数据
   - 提供to_dict()序列化方法
   - 支持扩展为数据库存储

### 使用示例

1. **添加日程**：
```json
{
  "tool": "add_schedule",
  "title": "团队周会",
  "start_time": "2023-10-16T10:00:00",
  "end_time": "2023-10-16T11:30:00",
  "reminder": "2023-10-16T09:45:00",
  "description": "每周产品迭代讨论"
}
```

2. **查询即将到来的日程**：
```json
{
  "tool": "query_schedules",
  "upcoming": true
}
```

3. **设置提醒**：
```json
{
  "tool": "set_reminder",
  "schedule_id": "1697443200-0",
  "reminder_time": "2023-10-16T09:30:00"
}
```

## 系统集成

1. **与mcp_pipe.py集成**：
```bash
python mcp_pipe.py schedule_manager.py
```

2. **AI交互示例**：
```
用户：请帮我安排下周二下午3点与客户的会议，提前30分钟提醒我
AI：好的，已为您创建日程：
    标题：客户会议
    时间：2023-10-17T15:00:00
    提醒：2023-10-17T14:30:00
```

## 扩展建议

1. **持久化存储**：
   - 添加SQLite或Redis存储支持
   - 实现自动保存/加载功能

2. **日历集成**：
   - 支持导出为ICS日历格式
   - 集成Google Calendar或Outlook

3. **高级提醒**：
   - 添加多种提醒方式（邮件、短信）
   - 支持重复日程（每天/每周/每月）

4. **自然语言处理**：
   - 添加NLP解析时间表达式
   - 支持"明天上午10点"等自然语言输入

此日程管理MCP工具提供了完整的日程管理功能，通过MCP协议与AI系统集成，使AI能够帮助用户高效管理时间安排。