Prefect

What is mcp-prefect?

mcp-prefect is an MCP (Model Context Protocol) server designed to bridge the gap between AI assistants and Prefect, a workflow orchestration and ELT/ETL tool. It allows AI assistants to understand and execute commands related to Prefect through natural language, simplifying the management of data pipelines and workflows.

How to use mcp-prefect?

To use mcp-prefect, you need to set up your Prefect API URL and optionally an API key as environment variables (PREFECT_API_URL and PREFECT_API_KEY). The project can then be run using Docker Compose, which will start both the MCP server and Prefect. Once connected, AI assistants can interpret natural language commands to interact with Prefect.

Key Features of mcp-prefect

mcp-prefect provides access to various Prefect APIs, offering comprehensive control over your workflows:

• Flow Management: List, retrieve, and delete flows.
• Flow Run Management: Create, monitor, and control flow runs.
• Deployment Management: Manage deployments and their schedules.
• Task Run Management: Monitor and control individual task runs.
• Work Queue Management: Create and manage work queues for task execution.
• Block Management: Access different block types and their associated documents.
• Variable Management: Create and manage variables within Prefect.
• Workspace Management: Get information about Prefect workspaces.

Use Cases of mcp-prefect

mcp-prefect is ideal for scenarios where natural language interaction with Prefect is beneficial. This includes:

• Automated Workflow Management: AI assistants can trigger, monitor, and manage Prefect workflows based on verbal or text commands.
• Simplified Operations: Non-technical users can interact with complex data pipelines without needing to write code.
• Enhanced Monitoring: Quickly get status updates on flow runs and deployments by asking simple questions.
• Integration with AI Platforms: Seamlessly integrate Prefect into AI-powered applications and chatbots.

FAQ about mcp-prefect

Q: What is Prefect? A: Prefect is an open-source workflow management system that helps data engineers and scientists build, run, and monitor data pipelines.

Q: What is an MCP server? A: An MCP (Model Context Protocol) server enables AI models to interact with external services and tools by translating natural language commands into API calls.

Q: Are all Prefect endpoints implemented? A: No, some endpoints are still under development. The project README indicates that several endpoints have yet to be implemented.

Q: How can I add new functions to mcp-prefect? A: To add a new function, you need to add it to the appropriate module in src/mcp_prefect and include it in the get_all_functions() list within that module. For new API types, you need to update APIType in enums.py, create a new module, and update main.py.

Prefect MCP Server

A Model Context Protocol (MCP) server implementation for Prefect, allowing AI assistants to interact with Prefect through natural language.

Features

This MCP server provides access to the following Prefect APIs:

• Flow Management: List, get, and delete flows
• Flow Run Management: Create, monitor, and control flow runs
• Deployment Management: Manage deployments and their schedules
• Task Run Management: Monitor and control task runs
• Work Queue Management: Create and manage work queues
• Block Management: Access block types and documents
• Variable Management: Create and manage variables
• Workspace Management: Get information about workspaces

Configuration

Set the following environment variables:

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)

Usage

Run the MCP server, and prefect:

docker compose up

Example Input

Once connected, an AI assistant can help users interact with Prefect using natural language. Examples:

• "Show me all my flows"
• "List all failed flow runs from yesterday"
• "Trigger the 'data-processing' deployment"
• "Pause the schedule for the 'daily-reporting' deployment"
• "What's the status of my last ETL flow run?"

Development

Several of the endpoints have yet to be implemented

Adding New Functions

To add a new function to an existing API:

1. Add the function to the appropriate module in src/mcp_prefect
2. Add the function to the get_all_functions() list in the module

To add a new API type:

1. Add the new type to APIType in enums.py
2. Create a new module in src/prefect/
3. Update main.py to include the new API type

Example usage:

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