Q Scheduler MCP

🚀 MCP Scheduler

MCP Scheduler是一个强大的任务调度服务器，它基于模型上下文协议（MCP）构建，可用于调度和管理各类自动化任务。

该项目是从 https://github.com/PhialsBasement/scheduler-mcp 分叉而来。

🚀 快速开始

MCP Scheduler是一个多功能的任务自动化系统，你可以使用它来调度和运行不同类型的任务，如执行系统命令、发起HTTP请求、生成AI内容以及设置提醒等。使用前请确保满足安装的前提条件，安装完成后即可根据自身需求配置并运行。

✨ 主要特性

• 多任务类型支持：支持执行Shell命令、发起API调用、生成AI内容以及显示桌面提醒。
• Cron调度：采用熟悉的Cron语法，可精确控制任务调度时间。
• 单次或定期执行：可选择让任务仅执行一次，或者按计划重复执行。
• 执行历史记录：可跟踪任务的成功和失败执行情况。
• 跨平台支持：可在Windows、macOS和Linux系统上运行。
• 交互式通知：提醒任务会以带有声音的桌面警报形式呈现。
• MCP集成：能与AI助手和其他MCP兼容的客户端无缝连接。
• 强大的错误处理：具备全面的日志记录和错误恢复功能。

📦 安装指南

前提条件

• Python 3.10 或更高版本
• uv（推荐的包管理器）

安装uv（推荐）

# 对于Mac/Linux系统
curl -LsSf https://astral.sh/uv/install.sh | sh

# 对于Windows系统（PowerShell）
powershell -c "irm https://astral.sh/uv/install.ps1 | iex"

安装uv后，请重启终端以确保命令可用。

使用uv进行项目设置（推荐）

# 克隆仓库
git clone https://github.com/yourusername/mcp-scheduler.git
cd mcp-scheduler

# 使用uv创建并激活虚拟环境
uv venv
source .venv/bin/activate  # 在Unix/MacOS系统上
# 或者
.venv\Scripts\activate     # 在Windows系统上

# 使用uv安装依赖项
uv pip install -r requirements.txt

标准pip安装（可选）

如果你更喜欢使用标准的pip：

# 克隆仓库
git clone https://github.com/yourusername/mcp-scheduler.git
cd mcp-scheduler

# 创建并激活虚拟环境
python -m venv .venv
source .venv/bin/activate  # 在Unix/MacOS系统上
# 或者
.venv\Scripts\activate     # 在Windows系统上

# 安装依赖项
pip install -r requirements.txt

💻 使用示例

运行服务器

# 首先激活虚拟环境
source .venv/bin/activate  # 在Unix/MacOS系统上
# 或者
.venv\Scripts\activate     # 在Windows系统上

# 使用默认设置运行（stdio传输）
uv run main.py

# 与AWS Q集成运行（推荐给Amazon Q用户）
uv run start_with_aws_q.py

# 以调试模式运行，获取详细日志
uv run main.py --debug

# 使用自定义配置文件运行
uv run main.py --config /path/to/config.json

服务器默认使用stdio传输，这非常适合与Amazon Q和其他MCP客户端集成。服务器会根据环境自动处理通信协议。

与Amazon Q、Claude Desktop或其他MCP客户端集成

Amazon Q集成

要将MCP Scheduler与Amazon Q一起使用：

1. 确保你已安装Amazon Q CLI。
2. 使用AWS Q模型补丁运行调度器：

# 启动与AWS Q集成的调度器
uv run start_with_aws_q.py

这将自动将调度器注册到Amazon Q，使你能够通过自然语言命令创建和管理任务。

示例命令：

• "创建一个定时任务，每天晚上10:30备份我的配置文件"
• "显示所有定时任务"
• "立即运行备份任务"

更多与Amazon Q的使用示例，请查看examples目录。

Claude Desktop集成

要将MCP Scheduler与Claude Desktop一起使用：

1. 确保你已安装Claude Desktop。
2. 打开Claude Desktop应用程序配置文件：
   • macOS：~/Library/Application Support/Claude/claude_desktop_config.json
   • Windows：%APPDATA%\Claude\claude_desktop_config.json
3. 如果文件不存在，请创建该文件，并添加你的服务器：

{
  "mcpServers": [
    {
      "type": "stdio",
      "name": "MCP Scheduler",
      "command": "python",
      "args": ["/path/to/your/mcp-scheduler/main.py"]
    }
  ]
}

或者，如果你使用的是FastMCP库，可以使用fastmcp实用工具：

# 在Claude Desktop中安装你的服务器
fastmcp install main.py --name "Task Scheduler"

命令行选项

--address        服务器地址（默认：localhost）
--port           服务器端口（默认：8080）
--transport      传输模式（stdio或sse）（默认：stdio）
--log-level      日志级别（默认：INFO）
--log-file       日志文件路径（默认：mcp_scheduler.log）
--db-path        SQLite数据库路径（默认：scheduler.db）
--config         JSON配置文件路径
--ai-model       用于AI任务的AI模型（默认：gpt-4o）
--version        显示版本信息并退出
--debug          启用调试模式，显示完整回溯信息
--fix-json       启用JSON修复功能，处理格式错误的消息

当与Amazon Q一起使用时，大多数这些选项会由start_with_aws_q.py脚本自动配置。

配置文件

你可以使用JSON配置文件代替命令行参数：

{
  "server": {
    "name": "mcp-scheduler",
    "version": "0.1.0",
    "address": "localhost",
    "port": 8080,
    "transport": "stdio"
  },
  "database": {
    "path": "scheduler.db"
  },
  "logging": {
    "level": "INFO",
    "file": "mcp_scheduler.log"
  },
  "scheduler": {
    "check_interval": 5,
    "execution_timeout": 300
  },
  "ai": {
    "model": "gpt-4o",
    "use_aws_q_model": true,
    "openai_api_key": "your-api-key"  // 仅在不使用AWS Q模型时需要
  }
}

当与Amazon Q一起使用时，use_aws_q_model应设置为true，并且不需要API密钥。

📚 详细文档

任务类型和限制

MCP Scheduler是一个应用级任务调度服务，而不是系统级定时任务管理器：

• 应用级任务：MCP Scheduler创建的任务存储在其自己的数据库中，只有在MCP Scheduler服务运行时才会执行。
• 非系统级：这些任务不是系统crontab或systemd定时器，不会在系统启动时自动运行。
• 服务依赖：如果MCP Scheduler服务停止，任务将不会执行。
• 用户权限：任务以运行MCP Scheduler的用户权限执行，而不是root权限。

如果你需要系统级定时任务（在系统启动时自动运行或需要root权限），请考虑：

1. 使用操作系统的crontab -e或systemctl直接创建系统级定时任务。
2. 创建一个MCP Scheduler任务，该任务执行脚本来管理系统级定时任务。

持久化和服务管理

为确保MCP Scheduler在系统重启后继续运行，你可以：

1. 将其设置为系统服务（使用systemd）。
2. 在用户登录时自动启动。
3. 在云环境中作为容器或服务运行。

MCP工具函数

MCP Scheduler提供以下工具：

任务管理

• list_tasks：获取所有定时任务。
• get_task：获取特定任务的详细信息。
• add_command_task：添加一个新的Shell命令任务。
• add_api_task：添加一个新的API调用任务。
• add_ai_task：添加一个新的AI任务。
• add_reminder_task：添加一个带有桌面通知的新提醒任务。
• update_task：更新现有任务。
• remove_task：删除一个任务。
• enable_task：启用一个已禁用的任务。
• disable_task：禁用一个活动任务。
• run_task_now：立即运行一个任务。

执行和监控

• get_task_executions：获取任务的执行历史记录。
• get_server_info：获取服务器信息。

Cron表达式指南

MCP Scheduler使用标准的Cron表达式进行调度。以下是一些示例：

• 0 0 * * * - 每天午夜执行。
• 0 */2 * * * - 每2小时执行一次。
• 0 9-17 * * 1-5 - 周一至周五，上午9点到下午5点，每小时执行一次。
• 0 0 1 * * - 每月第一天午夜执行。
• 0 0 * * 0 - 每周日午夜执行。

环境变量

调度器可以使用环境变量进行配置：

• MCP_SCHEDULER_NAME：服务器名称（默认：mcp-scheduler）
• MCP_SCHEDULER_VERSION：服务器版本（默认：0.1.0）
• MCP_SCHEDULER_ADDRESS：服务器地址（默认：localhost）
• MCP_SCHEDULER_PORT：服务器端口（默认：8080）
• MCP_SCHEDULER_TRANSPORT：传输模式（默认：stdio）
• MCP_SCHEDULER_LOG_LEVEL：日志级别（默认：INFO）
• MCP_SCHEDULER_LOG_FILE：日志文件路径
• MCP_SCHEDULER_DB_PATH：数据库路径（默认：scheduler.db）
• MCP_SCHEDULER_CHECK_INTERVAL：检查任务的频率（默认：5秒）
• MCP_SCHEDULER_EXECUTION_TIMEOUT：任务执行超时时间（默认：300秒）
• MCP_SCHEDULER_AI_MODEL：用于AI任务的OpenAI模型（默认：gpt-4o）
• MCP_SCHEDULER_USE_AWS_Q_MODEL：是否使用AWS Q模型进行AI任务（默认：false）
• OPENAI_API_KEY：OpenAI任务的API密钥（使用AWS Q模型时不需要）

通过不同方式使用MCP Scheduler

通过Amazon Q使用（推荐）

使用Amazon Q创建和管理任务非常简单，只需使用自然语言描述你想要的任务：

1. 创建命令任务：

   创建一个定时任务，每天晚上10:30备份我的数据库到/backups目录
   

2. 创建API调用任务：

   设置一个每6小时获取一次天气数据的任务
   

3. 创建AI任务：

   每周一早上9点生成一份上周销售数据的摘要报告
   

4. 创建提醒任务：

   每周二和周四上午9:30提醒我参加团队会议
   

5. 查看所有任务：

   显示所有定时任务
   

6. 立即运行任务：

   立即运行备份任务
   

通过编程API使用

如果你正在开发应用程序或脚本，可以通过编程方式与MCP Scheduler交互。以下是建立调用关系的简要指南：

1. 安装必要的依赖

# 使用uv安装（推荐）
uv pip install "mcp[client]>=1.4.0"

2. 建立连接并调用API

import asyncio
from mcp.client import StdioClient

async def main():
    # 启动MCP Scheduler作为子进程
    process_args = ["uv", "run", "/path/to/scheduler-mcp/main.py"]
    async with StdioClient.create_subprocess(process_args) as client:
        # 获取服务器信息
        server_info = await client.call("get_server_info")
        print(f"连接到 {server_info['name']} 版本 {server_info['version']}")
        
        # 列出所有任务
        tasks = await client.call("list_tasks")
        print(f"当前有 {len(tasks)} 个任务")
        
        # 添加一个命令任务
        cmd_task = await client.call(
            "add_command_task", 
            {
                "name": "系统状态检查",
                "schedule": "*/30 * * * *",  # 每30分钟
                "command": "vmstat > /tmp/vmstat_$(date +%Y%m%d_%H%M).log",
                "description": "记录系统状态",
                "do_only_once": False
            }
        )
        print(f"创建命令任务: {cmd_task['id']}")
        
        # 立即运行一个任务
        run_result = await client.call(
            "run_task_now", 
            {"task_id": cmd_task['id']}
        )
        print(f"任务执行结果: {run_result['execution']['status']}")

# 运行主函数
if __name__ == "__main__":
    asyncio.run(main())

3. 连接到已运行的MCP Scheduler

如果MCP Scheduler已经在HTTP模式下运行，可以使用SSE客户端连接：

import asyncio
from mcp.client import SseClient

async def connect_to_running_scheduler():
    async with SseClient("http://localhost:8080") as client:
        tasks = await client.call("list_tasks")
        print(f"当前有 {len(tasks)} 个任务")

asyncio.run(connect_to_running_scheduler())

4. 错误处理

import asyncio
from mcp.client import StdioClient
from mcp.errors import McpError

async def robust_scheduler_client():
    try:
        process_args = ["uv", "run", "/path/to/scheduler-mcp/main.py"]
        async with StdioClient.create_subprocess(process_args) as client:
            try:
                result = await client.call("list_tasks")
                return result
            except McpError as e:
                print(f"MCP API错误: {e}")
                return []
    except Exception as e:
        print(f"连接错误: {e}")
        return []

asyncio.run(robust_scheduler_client())

完整示例

查看 examples/api_client_example.py 获取完整的API使用示例，包括：

• 连接到MCP Scheduler服务。
• 创建、运行、更新和删除任务。
• 获取任务执行历史。
• 错误处理和异常管理。

# 运行示例
cd examples
./api_client_example.py

示例脚本

examples目录包含了可直接使用的脚本和配置，适用于常见用例：

• backup_mcp_config.sh：一个用于备份Amazon Q MCP配置文件的脚本，包含基于日期的命名和保留策略。

MCP工具发现

MCP Scheduler支持通过模型上下文协议进行自动工具发现：

Stdio模式（默认）

在stdio模式下运行时（默认模式），工具发现会通过MCP协议自动进行。这是与Amazon Q和其他支持stdio通信的MCP客户端一起使用的推荐模式。

# 在stdio模式下运行（默认）
uv run main.py

HTTP模式（可选）

如果你需要在HTTP模式下运行服务器，可以使用SSE传输，并通过知名端点访问模式：

# 使用HTTP服务器传输运行
uv run main.py --transport sse --port 8080

在HTTP模式下，服务器会暴露一个知名端点用于工具/模式的自动发现：

• 端点：/.well-known/mcp-schema.json（在HTTP端口 + 1上，例如，如果你的服务器在8080端口运行，模式在8081端口）
• 目的：允许客户端和AI助手自动发现所有可用的MCP工具及其参数。

你可以在以下地址访问模式：

http://localhost:8081/.well-known/mcp-schema.json

示例模式响应

{
  "tools": [
    {
      "name": "list_tasks",
      "description": "List all scheduled tasks.",
      "endpoint": "list_tasks",
      "method": "POST",
      "parameters": {
        "type": "object",
        "properties": {},
        "required": [],
        "additionalProperties": false
      }
    },
    {
      "name": "add_command_task",
      "description": "Add a new shell command task.",
      "endpoint": "add_command_task",
      "method": "POST",
      "parameters": {
        "type": "object",
        "properties": {
          "name": {"type": "string"},
          "schedule": {"type": "string"},
          "command": {"type": "string"},
          "description": {"type": "string"},
          "enabled": {"type": "boolean"},
          "do_only_once": {"type": "boolean"}
        },
        "required": ["name", "schedule", "command"],
        "additionalProperties": false
      }
    }
    // ... more tools ...
  ]
}

此模式是从注册的MCP工具自动生成的，始终反映当前服务器的功能。

开发

如果你想为MCP Scheduler做出贡献或进一步开发，以下是一些额外的命令：

# 安装用于开发的MCP SDK
uv pip install "mcp[cli]>=1.4.0"

# 或者使用FastMCP（替代实现）
uv pip install fastmcp

# 测试你的MCP服务器
# 使用MCP Inspector工具
mcp inspect --stdio -- uv run main.py

# 或者使用简单的MCP客户端
python -m mcp.client.stdio uv run main.py

# 运行测试
uv run -m pytest

# 代码检查
uv pip install flake8
flake8 mcp_scheduler/

📄 许可证

本项目采用MIT许可证 - 详情请参阅LICENSE文件。