任务调度服务

基于Model Context Protocol (MCP)构建的自动化任务调度服务,支持多种任务类型和灵活的定时设置。

  • by RadiumGu
  • Python
  • 本地部署
  • 4 星标
  • 更新于
在 GitHub 查看 radiumgu/q-scheduler-mcp

使用说明

MCP调度器

一个使用模型上下文协议(MCP)构建的健壮的任务调度服务器,用于调度和管理各种类型的自动化任务。

叉从https://github.com/PhialsBasement/scheduler-mcp

概述

MCP Scheduler是一个多功能的任务自动化系统,允许您安排和运行不同类型的任务:

  • shell命令:按计划执行系统命令
  • API调用:向外部服务发出HTTP请求
  • AI任务:通过OpenAI模型生成内容
  • 提醒事项:显示带声音的桌面通知

调度器使用cron表达式实现灵活的定时,并提供任务执行的完整历史记录。它基于模型上下文协议(MCP)构建,使其易于与AI助手和其他兼容MCP的客户端集成。

特性

  • 多种任务类型:支持shell命令、API调用、AI内容生成和桌面通知
  • Cron调度:用于精确调度控制的熟悉cron语法
  • 运行一次或重复运行:按计划只运行一次或重复运行任务的选项
  • 执行历史:跟踪成功和失败的任务执行
  • 交叉平台的:适用于Windows、macOS和Linux
  • 交互式通知:用于提醒任务的带声音的桌面警报
  • MCP集成:与人工智能助手和工具无缝连接
  • 稳健的错误处理:全面的日志记录和错误恢复

安装

先决条件

  • Python 3.10或更高版本
  • 紫外线 (推荐包管理器)

安装uv(推荐)

# For Mac/Linux
curl -LsSf https://astral.sh/uv/install.sh | sh

# For Windows (PowerShell)
powershell -c "irm https://astral.sh/uv/install.ps1 | iex"

安装uv后,重新启动终端以确保命令可用。

使用uv进行项目设置(推荐)

# Clone the repository
git clone https://github.com/yourusername/mcp-scheduler.git
cd mcp-scheduler

# Create and activate a virtual environment with uv
uv venv
source .venv/bin/activate  # On Unix/MacOS
# or
.venv\Scripts\activate     # On Windows

# Install dependencies with uv
uv pip install -r requirements.txt

标准管道安装(备选)

如果你更喜欢使用标准pip:

# Clone the repository
git clone https://github.com/yourusername/mcp-scheduler.git
cd mcp-scheduler

# Create and activate a virtual environment
python -m venv .venv
source .venv/bin/activate  # On Unix/MacOS
# or
.venv\Scripts\activate     # On Windows

# Install dependencies
pip install -r requirements.txt

用法

运行服务器

# Activate the virtual environment first
source .venv/bin/activate  # On Unix/MacOS
# or
.venv\Scripts\activate     # On Windows

# Run with default settings (stdio transport)
uv run main.py

# Run with AWS Q integration (recommended for Amazon Q users)
uv run start_with_aws_q.py

# Run with debug mode for detailed logging
uv run main.py --debug

# Run with custom configuration file
uv run main.py --config /path/to/config.json

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

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

亚马逊Q集成

要将MCP调度器与Amazon Q一起使用:

  1. 确保已安装Amazon Q CLI
  2. 使用AWS Q模型补丁运行调度程序:
# Start the scheduler with AWS Q integration
uv run start_with_aws_q.py

这将自动向Amazon Q注册调度程序,允许您通过自然语言命令创建和管理任务。

命令示例:

  • “创建一个计划任务,每天晚上10:30备份我的配置文件”
  • “显示我的所有计划任务”
  • “立即运行备份任务”

请参阅 examples 有关Amazon Q的更多使用示例的目录。

Claude桌面集成

要在Claude Desktop中使用MCP调度程序,请执行以下操作:

  1. 确保您已安装Claude Desktop
  2. 在以下位置打开您的Claude桌面应用程序配置:
    • macOS: ~/Library/Application Support/Claude/claude_desktop_config.json
    • 窗户: %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库):

# Install your server in Claude Desktop
fastmcp install main.py --name "Task Scheduler"

命令行选项

--address        Server address (default: localhost)
--port           Server port (default: 8080)
--transport      Transport mode (stdio or sse) (default: stdio)
--log-level      Logging level (default: INFO)
--log-file       Log file path (default: mcp_scheduler.log)
--db-path        SQLite database path (default: scheduler.db)
--config         Path to JSON configuration file
--ai-model       AI model to use for AI tasks (default: gpt-4o)
--version        Show version and exit
--debug          Enable debug mode with full traceback
--fix-json       Enable JSON fixing for malformed messages

当与Amazon Q一起使用时,这些选项中的大多数都是由 start_with_aws_q.py 脚本。 --config JSON配置文件的路径 --ai模型用于ai任务的ai模型(默认:gpt-4o) --version显示版本并退出 --debug启用带有完整回溯的调试模式 --fix json为格式错误的消息启用json修复


### Configuration File

You can use a JSON configuration file instead of command-line arguments:

```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"  // Only needed if not using AWS Q model
  }
}

使用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调度器提供以下工具:

任务管理

  • 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调度器使用标准的cron表达式进行调度。以下是一些示例:

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

环境变量

可以使用环境变量配置调度器:

  • MCP_SCHEDULER_NAME:服务器名称(默认值:mcp调度程序)
  • 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等MCP客户端的自然语言交互,或者通过编程方式直接调用API。

通过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调度器支持通过模型上下文协议自动发现工具:

标准模式(默认)

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

# Run in stdio mode (default)
uv run main.py

HTTP模式(可选)

如果需要在HTTP模式下运行服务器,可以使用SSE传输并通过众所周知的端点访问模式:

# Run with HTTP server transport
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调度器,这里有一些额外的命令:

# Install the MCP SDK for development
uv pip install "mcp[cli]>=1.4.0"

# Or for FastMCP (alternative implementation)
uv pip install fastmcp

# Testing your MCP server
# With the MCP Inspector tool
mcp inspect --stdio -- uv run main.py

# Or with a simple MCP client
python -m mcp.client.stdio uv run main.py

# Running tests
uv run -m pytest

# Linting
uv pip install flake8
flake8 mcp_scheduler/

许可证

此项目根据MIT许可证获得许可-有关详细信息,请参阅许可证文件。