MCP 消息提醒服务

基于 Model Context Protocol (MCP) 的消息发送和定时提醒服务，使用官方 MCP Python SDK 实现，支持 Telegram、飞书等多种消息平台。

✨ 功能特点

🔌 MCP 协议: 基于官方 MCP Python SDK，完全符合 MCP 规范
🌐 多传输支持: stdio、HTTP、SSE、both 四种传输方式，适应不同客户端
🚀 FastMCP: 使用官方 FastMCP 提供高性能 API 接口
📱 多平台消息: 支持 Telegram、飞书等多种消息平台
⏰ 定时任务: 基于 Cron 表达式的定时消息发送
🎯 最佳实践: 符合 Python 项目最佳实践和 MCP Server 规范
🐳 容器化部署: 完整的 Docker 支持，一键部署
📝 完善日志: 结构化日志记录，支持日志轮转
🔄 自动重试: 智能错误处理和重试机制
⚡ 异步处理: 高性能异步架构，支持并发发送
🛠️ 开发友好: 完整的开发工具链和测试支持

快速开始

1. 配置 Telegram（可选）

在 Telegram 中找到 @BotFather
发送 /newbot 命令创建新机器人
按提示设置机器人名称和用户名
获取 Bot Token
在 Telegram 中找到 @userinfobot
发送任意消息获取你的 Chat ID

2. 配置飞书（可选）

在飞书群聊中添加自定义机器人
获取 Webhook URL
如果配置了签名验证，获取签名密钥

3. 配置环境变量

复制 env.example 文件并填入你的配置：

cp env.example .env

编辑 .env 文件：

# Telegram 配置（可选） TELEGRAM_BOT_TOKEN=你的机器人Token TELEGRAM_CHAT_ID=你的Chat ID # 飞书配置（可选） FEISHU_WEBHOOK_URL=https://open.feishu.cn/open-apis/bot/v2/hook/your_webhook_url FEISHU_SECRET=your_feishu_secret_here # 可选

注意: 至少需要配置一个消息发送平台（Telegram 或飞书），程序才能正常运行。

4. 本地运行

本地模式（用于 Cursor MCP）

# 安装依赖 pip install -r requirements.txt # 设置环境变量 export TELEGRAM_BOT_TOKEN="你的机器人Token" export TELEGRAM_CHAT_ID="你的Chat ID" export SERVER_MODE="local" # 运行 MCP 服务器 python main.py

HTTP API 模式

# 安装依赖 pip install -r requirements.txt # 设置环境变量 export TELEGRAM_BOT_TOKEN="你的机器人Token" export TELEGRAM_CHAT_ID="你的Chat ID" export SERVER_MODE="http" export SERVER_HOST="0.0.0.0" export SERVER_PORT="8000" # 运行 HTTP API 服务器 python main.py

SSE 模式

# 安装依赖 pip install -r requirements.txt # 设置环境变量 export TELEGRAM_BOT_TOKEN="你的机器人Token" export TELEGRAM_CHAT_ID="你的Chat ID" export SERVER_MODE="sse" export SERVER_HOST="0.0.0.0" export SERVER_PORT="8000" # 运行 SSE 服务器 python main.py

双模式（HTTP + SSE）

# 安装依赖 pip install -r requirements.txt # 设置环境变量 export TELEGRAM_BOT_TOKEN="你的机器人Token" export TELEGRAM_CHAT_ID="你的Chat ID" export SERVER_MODE="both" export SERVER_HOST="0.0.0.0" export SERVER_PORT="8000" # 同时运行 HTTP 和 SSE 服务器 # HTTP: http://localhost:8000/mcp # SSE: http://localhost:8001/sse python main.py

测试 HTTP API

# 获取工具列表 curl http://localhost:8000/tools # 发送消息 curl -X POST http://localhost:8000/tools/send_message \ -H "Content-Type: application/json" \ -d '{"provider": "telegram", "message": "测试消息"}' # 测试连接 curl -X POST http://localhost:8000/tools/test_connection

5. Docker 运行

# 构建镜像 docker build -t telegram-reminder . # 运行容器 docker run -d \ --name telegram-reminder \ --env-file .env \ --restart unless-stopped \ telegram-reminder

6. Docker Compose 运行

创建 docker-compose.yml 文件：

version: '3.8' services: telegram-reminder: build: . container_name: telegram-reminder env_file: - .env restart: unless-stopped volumes: - ./logs:/app/logs

运行：

docker-compose up -d

配置说明

环境变量

变量名	说明	必需
`TELEGRAM_BOT_TOKEN`	Telegram Bot Token	否*
`TELEGRAM_CHAT_ID`	接收消息的 Chat ID	否*
`FEISHU_WEBHOOK_URL`	飞书机器人 Webhook URL	否*
`FEISHU_SECRET`	飞书机器人签名密钥	否

*至少需要配置 Telegram 或飞书中的一个平台

自定义消息

如需修改提醒消息，可以编辑 telegram_reminder.py 文件中的 self.message 变量：

self.message = "你的自定义消息 💊"

日志

脚本会生成详细的日志文件 telegram_reminder.log，包含：

服务启动/停止信息
下次提醒时间计算
消息发送状态
错误信息和重试记录

故障排除

常见问题

消息发送失败
- 检查 Bot Token 是否正确
- 确认 Chat ID 是否正确
- 确保机器人已添加到对话中
时间不准确
- 脚本使用北京时间（UTC+8）
- 容器时区已设置为 Asia/Shanghai
容器无法启动
- 检查环境变量是否正确设置
- 查看容器日志：docker logs telegram-reminder

查看日志

# 查看容器日志 docker logs telegram-reminder # 查看应用日志文件 docker exec telegram-reminder cat /app/telegram_reminder.log

🏗️ 系统架构

mcp_reminder/ ├── server/ # MCP 服务器 │ ├── simple_mcp_server.py # 简化的 MCP 服务器（基于官方 SDK） │ ├── fast_mcp_server.py # FastMCP HTTP 服务器 │ └── http_server.py # 传统 HTTP 模式服务器 ├── tools/ # MCP 工具 │ ├── message_tools.py # 消息发送工具 │ └── scheduler_tools.py # 定时任务工具 ├── providers/ # 消息提供商 │ ├── message_sender.py # 消息发送管理器 │ ├── telegram.py # Telegram 提供商 │ └── feishu.py # 飞书提供商 ├── config/ # 配置管理 │ └── settings.py # 配置类 ├── utils/ # 工具模块 │ └── logger.py # 日志工具 └── __init__.py # 包初始化

核心组件

SimpleMCPServer: 基于官方 MCP SDK 的简化服务器，处理工具调用（本地模式）
FastMCPServer: 基于官方 FastMCP 的 HTTP 服务器，提供高性能 API 接口
HTTPServer: 传统 HTTP 模式服务器，提供 RESTful API
MessageTools: 消息发送工具，支持单平台和全平台发送
SchedulerTools: 定时任务工具，基于 Cron 表达式
MessageSender: 消息发送管理器，管理多个提供商

🛠️ 开发指南

本地开发

# 安装依赖 make install # 运行测试 make test # 代码检查 make lint # 代码格式化 make format # 本地运行（HTTP 模式） make run-http # 本地运行（本地模式） make run-local

添加新的消息提供商

在 reminder/providers/ 目录下创建新的提供商文件
继承 MessageProvider 基类
实现 send_message 和 is_configured 方法
在 MessageSender 中添加初始化逻辑

示例：

from .base import MessageProvider class CustomProvider(MessageProvider): def __init__(self, api_key: str): super().__init__("Custom") self.api_key = api_key async def send_message(self, message: str, **kwargs) -> bool: # 实现发送逻辑 pass def is_configured(self) -> bool: return bool(self.api_key)

🚀 部署

使用 Makefile 一键部署

make deploy

手动部署

# 构建镜像 docker build -t reminder-system . # 运行容器 docker run -d \ --name reminder-system \ --env-file .env \ --restart unless-stopped \ reminder-system

📊 监控和日志

日志文件: 支持日志轮转，默认保留 5 个备份文件
日志级别: 可通过 LOG_LEVEL 环境变量配置
连接测试: 使用 python main.py test 测试所有提供商连接

🧪 测试

# 运行所有测试 make test # 运行特定测试 python -m pytest tests/test_config.py -v # 生成覆盖率报告 python -m pytest --cov=reminder tests/

技术实现

异步处理: 使用 asyncio 和 aiohttp 实现高性能异步处理
时区处理: 正确处理北京时间时区
错误处理: 完善的异常处理和重试机制
日志记录: 结构化日志记录，支持日志轮转
配置管理: 基于 dataclass 的类型安全配置
测试覆盖: 完整的单元测试和集成测试

许可证

MIT License

MCP Reminder Service