使用说明
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
