Prefect MCP 服务器

一个用于 Prefect 的模型上下文协议（MCP）服务器实现，允许 AI 助手通过自然语言与 Prefect 进行交互。

特性

此 MCP 服务器提供了对以下 Prefect API 的访问：

• 流管理：列出、获取和删除流
• 流运行管理：创建、监控和控制流运行
• 部署管理：管理部署及其计划
• 任务运行管理：监控和控制任务运行
• 工作队列管理：创建和管理工作队列
• 块管理：访问块类型和文档
• 变量管理：创建和管理变量
• 工作区管理：获取有关工作区的信息

配置

设置以下环境变量：

export PREFECT_API_URL="http://localhost:4200/api"  # URL of your Prefect API
export PREFECT_API_KEY="your_api_key"               # Your Prefect API key (if using Prefect Cloud)

使用方法

运行 MCP 服务器及 prefect：

docker compose up

示例输入

一旦连接成功，AI 助手可以帮助用户使用自然语言与 Prefect 交互。例如：

• "显示我所有的流"
• "列出昨天所有失败的流运行"
• "触发 'data-processing' 部署"
• "暂停 'daily-reporting' 部署的时间表"
• "我的最后一次 ETL 流运行的状态是什么？"

开发

有几个端点尚未实现

添加新功能

要向现有 API 添加新功能：

1. 将函数添加到 src/mcp_prefect 中相应的模块
2. 将函数添加到该模块中的 get_all_functions() 列表

要添加新的 API 类型：

1. 在 enums.py 中将新类型添加到 APIType
2. 在 src/prefect/ 中创建一个新的模块
3. 更新 main.py 以包含新的 API 类型

示例用法：

{
  "mcpServers": {
    "mcp-prefect": {
      "command": "mcp-prefect",
      "args": [
        "--transport", "sse"
      ],
      "env": {
        "PYTHONPATH": "/path/to/your/project/directory"
      },
      "cwd": "/path/to/your/project/directory"
    }
  }
}