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/
• 在仓库中创建一个问题