关键词

Prefect工作流自动化AI助手自然语言接口服务器

使用说明

项目简介

Prefect MCP Server 是一个 Model Context Protocol (MCP) 服务器的实现，它允许 AI 助手通过标准化的 MCP 协议与 Prefect 工作流自动化平台进行交互。该服务器提供了一系列工具，使 AI 能够查询、控制和管理 Prefect 的 Flows, Flow Runs, Deployments, Task Runs, Work Queues, Blocks, Variables 和 Workspaces。

主要功能点

  • 工作流管理 (Flow Management): 列出、获取和删除 Prefect Flows。
  • 工作流运行管理 (Flow Run Management): 创建、监控和控制 Flow Runs。
  • 部署管理 (Deployment Management): 管理 Deployments 及其 Schedules。
  • 任务运行管理 (Task Run Management): 监控和控制 Task Runs。
  • 工作队列管理 (Work Queue Management): 创建和管理 Work Queues。
  • Block 管理 (Block Management): 访问和管理 Prefect Blocks。
  • 变量管理 (Variable Management): 创建和管理 Variables。
  • 工作空间管理 (Workspace Management): 获取 Workspace 信息。
  • 健康检查 (Health Check): 提供服务器健康状态检查。

安装步骤

  1. 环境准备: 确保已安装 Docker 和 Docker Compose。
  2. 克隆仓库: 从 GitHub 克隆 'mcp-prefect' 仓库到本地。 
    git clone https://github.com/allen-munsch/mcp-prefect.git
cd mcp-prefect
  3. 配置环境变量: 根据您的 Prefect 环境配置以下环境变量。 
    export PREFECT_API_URL="http://localhost:4200/api"  # 您的 Prefect API URL
export PREFECT_API_KEY="your_api_key"               # 您的 Prefect API 密钥 (如果使用 Prefect Cloud)
  4. 启动服务器: 使用 Docker Compose 启动 MCP 服务器和 Prefect (如果需要)。 
    docker compose up

服务器配置

MCP 客户端需要以下配置信息连接到 Prefect MCP Server。请注意，服务器通过 Docker Compose 启动，客户端需要连接到运行中的服务器实例。

{
  "serverName": "Prefect MCP",  // MCP 服务器名称，来自 server.py 中的 FastMCP 初始化
  "command": "python -m mcp_prefect.main", // MCP 服务器启动命令，使用 Python 模块方式运行
  "args": ["--transport", "sse"] // 启动参数，指定使用 SSE 传输协议
}

配置参数说明:

  • 'serverName': 指定服务器的名称，客户端可以通过此名称识别服务器类型。
  • 'command': 指定启动 MCP 服务器的命令。这里使用 Python 模块方式 'python -m mcp_prefect.main' 运行 'main.py' 文件，确保 Python 环境中 'mcp_prefect' 包已正确安装。
  • 'args': 启动参数列表，'"--transport sse"' 指定服务器使用 Server-Sent Events (SSE) 传输协议，这通常是推荐的客户端连接方式。

基本使用方法

  1. 连接服务器: 在 MCP 客户端中，配置上述服务器配置信息，建立与 Prefect MCP Server 的连接。
  2. 调用工具: 客户端可以使用自然语言指令，通过 MCP 协议调用服务器提供的工具，例如：
    • "Show me all my flows" (对应 'get_flows' 工具)
    • "List all failed flow runs from yesterday" (可能对应 'get_flow_runs' 工具，并需要客户端进行日期过滤)
    • "Trigger the 'data-processing' deployment" (对应 'create_flow_run_from_deployment' 工具)
    • "Pause the schedule for the 'daily-reporting' deployment" (对应 'pause_deployment_schedule' 工具)
    • "What's the status of my last ETL flow run?" (可能对应 'get_flow_runs' 或 'get_task_runs' 工具，并需要客户端进行状态和名称过滤)
  3. 接收响应: 服务器会将 Prefect 的相关信息以 JSON-RPC 响应的形式返回给客户端，客户端需要解析这些响应并呈现给用户。

注意: 部分端点可能尚未完全实现，具体功能可用性请参考仓库的开发进度。

信息

分类

商业系统

访问项目