Datapilot MCP Server

🚀 DataPilot MCP Server

DataPilot MCP 服務器是一個全面的模型上下文協議（MCP）服務器，藉助自然語言和人工智能與 Snowflake 進行交互。它基於 FastMCP 2.0 構建，並集成了 OpenAI，能在人工智能的引導下幫助你更好地管理數據。

🚀 快速開始

如果你想開始使用 DataPilot MCP Server，可按以下步驟操作：

1. 克隆並設置項目：

   git clone <repository-url>
   cd datapilot
   python -m venv venv
   source venv/bin/activate  # 在 Windows 系統上使用：venv\Scripts\activate
   

2. 安裝依賴項：

   pip install -r requirements.txt
   

3. 配置環境變量：

   cp env.template .env
   # 編輯 .env 文件並填入你的憑證信息
   

✨ 主要特性

🗄️ 核心數據庫操作

• execute_sql - 執行 SQL 查詢並獲取結果
• list_databases - 列出所有可訪問的數據庫
• list_schemas - 列出數據庫中的模式
• list_tables - 列出數據庫/模式中的表
• describe_table - 獲取詳細的表列信息
• get_table_sample - 從表中檢索樣本數據

🏭 倉庫管理

• list_warehouses - 列出所有可用的倉庫
• get_warehouse_status - 獲取當前倉庫、數據庫和模式的狀態

🤖 人工智能驅動的功能

• natural_language_to_sql - 將自然語言問題轉換為 SQL 查詢
• analyze_query_results - 對查詢結果進行人工智能分析
• suggest_query_optimizations - 獲取 SQL 查詢的優化建議
• explain_query - 用通俗易懂的英語解釋 SQL 查詢
• generate_table_insights - 由人工智能生成關於表數據的見解

📊 資源（數據訪問）

• snowflake://databases - 訪問數據庫列表
• snowflake://schemas/{database} - 訪問模式列表
• snowflake://tables/{database}/{schema} - 訪問表列表
• snowflake://table/{database}/{schema}/{table} - 訪問表詳細信息

📝 提示（模板）

• sql_analysis_prompt - SQL 分析模板
• data_exploration_prompt - 數據探索模板
• sql_optimization_prompt - 查詢優化模板

📦 安裝指南

環境變量

創建一個 .env 文件，並進行如下配置：

# 必需：Snowflake 連接
# 賬戶示例：
# - ACCOUNT-LOCATOR.snowflakecomputing.com（推薦）
# - ACCOUNT-LOCATOR.region.cloud
# - organization-account_name
SNOWFLAKE_ACCOUNT=ACCOUNT-LOCATOR.snowflakecomputing.com
SNOWFLAKE_USER=your_username
SNOWFLAKE_PASSWORD=your_password

# 可選：默認 Snowflake 上下文
SNOWFLAKE_WAREHOUSE=your_warehouse_name
SNOWFLAKE_DATABASE=your_database_name
SNOWFLAKE_SCHEMA=your_schema_name
SNOWFLAKE_ROLE=your_role_name

# 必需：OpenAI API
OPENAI_API_KEY=your_openai_api_key
OPENAI_MODEL=gpt-4  # 可選，默認為 gpt-4

Snowflake 賬戶設置

1. 獲取你的 Snowflake 賬戶標識符 - 支持多種格式：
   • 推薦：ACCOUNT-LOCATOR.snowflakecomputing.com（例如，SCGEENJ-UR66679.snowflakecomputing.com）
   • 區域：ACCOUNT-LOCATOR.region.cloud（例如，xy12345.us-east-1.aws）
   • 舊版：organization-account_name
2. 確保你的用戶具有適當的權限：
   • 對倉庫、數據庫和模式具有 USAGE 權限
   • 對錶具有 SELECT 權限以進行查詢
   • 對列出對象具有 SHOW 權限

💻 使用示例

運行服務器

方法 1：直接執行

python -m src.main

方法 2：使用 FastMCP CLI

fastmcp run src/main.py

方法 3：開發模式下自動重新加載

fastmcp dev src/main.py

連接到 MCP 客戶端

Claude 桌面版

在你的 Claude 桌面版配置中添加以下內容：

{
  "mcpServers": {
    "datapilot": {
      "command": "python",
      "args": ["-m", "src.main"],
      "cwd": "/path/to/datapilot",
      "env": {
        "SNOWFLAKE_ACCOUNT": "your_account",
        "SNOWFLAKE_USER": "your_user",
        "SNOWFLAKE_PASSWORD": "your_password",
        "OPENAI_API_KEY": "your_openai_key"
      }
    }
  }
}

使用 FastMCP 客戶端

from fastmcp import Client

async def main():
    async with Client("python -m src.main") as client:
        # 列出數據庫
        databases = await client.call_tool("list_databases")
        print("Databases:", databases)
        
        # 自然語言轉 SQL
        result = await client.call_tool("natural_language_to_sql", {
            "question": "Show me the top 10 customers by revenue",
            "database": "SALES_DB",
            "schema": "PUBLIC"
        })
        print("Generated SQL:", result)

具體使用示例

1. 自然語言查詢

# 用自然語言提出問題
question = "What are the top 5 products by sales volume last month?"
sql = await client.call_tool("natural_language_to_sql", {
    "question": question,
    "database": "SALES_DB",
    "schema": "PUBLIC"
})
print(f"Generated SQL: {sql}")

2. 執行並分析

# 執行查詢並進行人工智能分析
analysis = await client.call_tool("analyze_query_results", {
    "query": "SELECT product_name, SUM(quantity) as total_sales FROM sales GROUP BY product_name ORDER BY total_sales DESC LIMIT 10",
    "results_limit": 100,
    "analysis_type": "summary"
})
print(f"Analysis: {analysis}")

3. 表見解

# 獲取關於表的人工智能見解
insights = await client.call_tool("generate_table_insights", {
    "table_name": "SALES_DB.PUBLIC.CUSTOMERS",
    "sample_limit": 50
})
print(f"Table insights: {insights}")

4. 查詢優化

# 獲取優化建議
optimizations = await client.call_tool("suggest_query_optimizations", {
    "query": "SELECT * FROM large_table WHERE date_column > '2023-01-01'"
})
print(f"Optimization suggestions: {optimizations}")

📚 詳細文檔

架構

┌─────────────────┐    ┌─────────────────┐    ┌─────────────────┐
│   MCP Client    │    │   FastMCP       │    │   Snowflake     │
│   (Claude/etc)  │◄──►│   Server        │◄──►│   Database      │
└─────────────────┘    └─────────────────┘    └─────────────────┘
                                │
                                ▼
                       ┌─────────────────┐
                       │   OpenAI API    │
                       │   (GPT-4)       │
                       └─────────────────┘

項目結構

datapilot/
├── src/
│   ├── __init__.py
│   ├── main.py              # 主 FastMCP 服務器
│   ├── models.py            # Pydantic 數據模型
│   ├── snowflake_client.py  # Snowflake 連接和操作
│   └── openai_client.py     # OpenAI 集成
├── requirements.txt         # Python 依賴項
├── env.template            # 環境變量模板
└── README.md              # 本文件

🔧 技術細節

開發

添加新工具

1. 在 src/main.py 中定義你的工具函數：

@mcp.tool()
async def my_new_tool(param: str, ctx: Context) -> str:
    """工具功能描述"""
    await ctx.info(f"Processing: {param}")
    # 你的邏輯代碼
    return "result"

2. 添加適當的錯誤處理和日誌記錄
3. 使用 FastMCP 開發模式進行測試：fastmcp dev src/main.py

添加新資源

@mcp.resource("snowflake://my-resource/{param}")
async def my_resource(param: str) -> Dict[str, Any]:
    """資源描述"""
    # 你的邏輯代碼
    return {"data": "value"}

故障排除

常見問題

1. 連接錯誤
   • 驗證 .env 文件中的 Snowflake 憑證
   • 檢查網絡連接
   • 確保用戶具有所需的權限
2. OpenAI 錯誤
   • 驗證 OPENAI_API_KEY 是否設置正確
   • 檢查 API 配額和賬單
   • 確保模型名稱正確
3. 導入錯誤
   • 激活虛擬環境
   • 安裝所有依賴項：pip install -r requirements.txt
   • 從項目根目錄運行

日誌記錄

啟用調試日誌：

LOG_LEVEL=DEBUG

📄 許可證

本項目採用 MIT 許可證。

支持

如果你遇到問題或有疑問，可以：

• 查看故障排除部分
• 查看 FastMCP 文檔：https://gofastmcp.com/
• 在倉庫中創建一個問題