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（可选）

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

2. 配置飞书（可选）

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

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，包含：

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

故障排除

常见问题

1. 消息发送失败

   • 检查 Bot Token 是否正确
   • 确认 Chat ID 是否正确
   • 确保机器人已添加到对话中

2. 时间不准确

   • 脚本使用北京时间（UTC+8）
   • 容器时区已设置为 Asia/Shanghai

3. 容器无法启动

   • 检查环境变量是否正确设置
   • 查看容器日志：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

添加新的消息提供商

1. 在 reminder/providers/ 目录下创建新的提供商文件
2. 继承 MessageProvider 基类
3. 实现 send_message 和 is_configured 方法
4. 在 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