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 文件。