小智AI - 日程管理MCP工具实现
# 日程管理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 =
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能够帮助用户高效管理时间安排。
页:
[1]