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一起使用:
- 确保已安装Amazon Q CLI
- 使用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调度程序,请执行以下操作:
- 确保您已安装Claude Desktop
- 在以下位置打开您的Claude桌面应用程序配置:
- macOS: ~/Library/Application Support/Claude/claude_desktop_config.json - 窗户: %APPDATA%\Claude\claude_desktop_config.json
- 如果文件不存在,请创建该文件,并添加您的服务器:
{
"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:
{ "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`直接创建系统级定时任务
1. 创建一个MCP Scheduler任务,该任务执行脚本来管理系统级定时任务
### 持久化和服务管理
为确保MCP Scheduler在系统重启后继续运行,您可以:
1. 将其设置为系统服务(使用systemd)
1. 在用户登录时自动启动
1. 在云环境中作为容器或服务运行
## 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目录
1. **创建API调用任务**:
设置一个每6小时获取一次天气数据的任务
1. **创建AI任务**:
每周一早上9点生成一份上周销售数据的摘要报告
1. **创建提醒任务**:
每周二和周四上午9:30提醒我参加团队会议
1. **查看所有任务**:
显示所有定时任务
1. **立即运行任务**:
立即运行备份任务
### 通过编程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许可证获得许可-有关详细信息,请参阅许可证文件。