Datapilot Mcp Server

数据库 Python

🚀 DataPilot MCP Server

DataPilot MCP 服务器是一个全面的模型上下文协议（MCP）服务器，借助自然语言和人工智能与 Snowflake 进行交互。它基于 FastMCP 2.0 构建，并集成了 OpenAI，能在人工智能的引导下帮助你更好地管理数据。

🚀 快速开始

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

克隆并设置项目：

git clone 
cd datapilot
python -m venv venv
source venv/bin/activate  # 在 Windows 系统上使用：venv\Scripts\activate

安装依赖项：
```
pip install -r requirements.txt
```

配置环境变量：

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 账户设置

获取你的 Snowflake 账户标识符 - 支持多种格式：
- 推荐：ACCOUNT-LOCATOR.snowflakecomputing.com（例如，SCGEENJ-UR66679.snowflakecomputing.com）
- 区域：ACCOUNT-LOCATOR.region.cloud（例如，xy12345.us-east-1.aws）
- 旧版：organization-account_name
确保你的用户具有适当的权限：
- 对仓库、数据库和模式具有 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              # 本文件

🔧 技术细节

开发

添加新工具

在 src/main.py 中定义你的工具函数：

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

添加适当的错误处理和日志记录
使用 FastMCP 开发模式进行测试：fastmcp dev src/main.py

添加新资源

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

故障排除

常见问题

连接错误
- 验证 .env 文件中的 Snowflake 凭证
- 检查网络连接
- 确保用户具有所需的权限
OpenAI 错误
- 验证 OPENAI_API_KEY 是否设置正确
- 检查 API 配额和账单
- 确保模型名称正确
导入错误
- 激活虚拟环境
- 安装所有依赖项：pip install -r requirements.txt
- 从项目根目录运行

日志记录

启用调试日志：

LOG_LEVEL=DEBUG

📄 许可证

本项目采用 MIT 许可证。

支持

如果你遇到问题或有疑问，可以：

查看故障排除部分
查看 FastMCP 文档：https://gofastmcp.com/
在仓库中创建一个问题

0 条评论
分类：数据库