Q Scheduler MCP

🚀 MCP Scheduler

MCP Scheduler 是一个基于模型上下文协议（MCP）构建的强大任务调度服务器，用于调度和管理各种类型的自动化任务。它支持多种任务类型，可灵活定时，并能与 AI 助手等 MCP 兼容客户端无缝集成。

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

🚀 快速开始

MCP Scheduler 是一个多功能的任务自动化系统，可调度和运行不同类型的任务，包括执行系统命令、发起 API 调用、生成 AI 内容以及显示桌面提醒等。它使用 cron 表达式实现灵活的定时功能，并提供完整的任务执行历史记录。

✨ 主要特性

• 多任务类型：支持 shell 命令、API 调用、AI 内容生成和桌面通知。
• Cron 调度：使用熟悉的 cron 语法进行精确的调度控制。
• 单次或重复运行：可选择任务仅运行一次或按计划重复运行。
• 执行历史：跟踪任务的成功和失败执行情况。
• 跨平台：可在 Windows、macOS 和 Linux 上运行。
• 交互式通知：提醒任务的桌面警报并带有声音。
• MCP 集成：与 AI 助手和工具无缝连接。
• 强大的错误处理：全面的日志记录和错误恢复功能。

📦 安装指南

前提条件

• 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，允许您通过自然语言命令创建和管理任务。

示例命令：

• "Create a scheduled task to backup my config file every night at 10:30 PM"
• "Show me all my scheduled tasks"
• "Run the backup task now"

更多与 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 模型时不需要）。

通过 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
      }
    }
    // ... 更多工具 ...
  ]
}

此模式是从注册的 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 文件。