数据助理MCP服务器
  ](https://www.python.org/downloads/)    
在人工智能指导下导航您的数据。一个全面的模型上下文协议（MCP）服务器，用于使用自然语言和人工智能与Snowflake交互。基于FastMCP 2.0和OpenAI集成构建。
特性
🗄️ 核心数据库操作
- execute_sql -执行带有结果的SQL查询
- 列表_数据库 -列出所有可访问的数据库
- list_schemas -列出数据库中的架构
- list_tables -列出数据库/架构中的表
- describe_table -获取详细的表列信息
- get_table_sample -从表中检索示例数据
🏭 仓库管理
- 列表_宿舍 -列出所有可用仓库
- get_warehouse_status -获取当前仓库、数据库和架构状态
🤖 AI驱动的功能
- 自然语言to.sql -将自然语言问题转换为SQL查询
- 分析查询结果 -基于人工智能的查询结果分析
- 建议查询优化 -获取SQL查询的优化建议
- 解释查询 -SQL查询的简明英语解释
- generate_table_insights -人工智能生成的表格数据洞察
📊 资源（数据访问）
snowflake://databases-访问数据库列表snowflake://schemas/{database}-访问架构列表snowflake://tables/{database}/{schema}-访问表列表snowflake://table/{database}/{schema}/{table}-访问表详细信息
📝 提示（模板）
- sql_analysis_prompt -SQL分析模板
- 数据_探索_尝试 -数据探索模板
- sql_optimization_prompt -查询优化模板
安装
- 克隆并设置项目：
git clone
cd datapilot
python -m venv venv
source venv/bin/activate # On Windows: venv\Scripts\activate- 安装依赖项：
pip install -r requirements.txt- 配置环境变量：
cp env.template .env
# Edit .env with your credentials配置
环境变量
创建一个 .env 具有以下配置的文件：
# Required: Snowflake Connection
# Account examples:
# - ACCOUNT-LOCATOR.snowflakecomputing.com (recommended)
# - ACCOUNT-LOCATOR.region.cloud
# - organization-account_name
SNOWFLAKE_ACCOUNT=ACCOUNT-LOCATOR.snowflakecomputing.com
SNOWFLAKE_USER=your_username
SNOWFLAKE_PASSWORD=your_password
# Optional: Default Snowflake Context
SNOWFLAKE_WAREHOUSE=your_warehouse_name
SNOWFLAKE_DATABASE=your_database_name
SNOWFLAKE_SCHEMA=your_schema_name
SNOWFLAKE_ROLE=your_role_name
# Required: OpenAI API
OPENAI_API_KEY=your_openai_api_key
OPENAI_MODEL=gpt-4 # Optional, defaults to gpt-4Snowflake帐户设置
- 获取您的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 Desktop配置中：
{
"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:
# List databases
databases = await client.call_tool("list_databases")
print("Databases:", databases)
# Natural language to 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.自然语言查询
# Ask a question in natural language
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.执行和分析
# Execute a query and get AI analysis
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.表见解
# Get AI-powered insights about a table
insights = await client.call_tool("generate_table_insights", {
"table_name": "SALES_DB.PUBLIC.CUSTOMERS",
"sample_limit": 50
})
print(f"Table insights: {insights}")4.查询优化
# Get optimization suggestions
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 # Main FastMCP server
│ ├── models.py # Pydantic data models
│ ├── snowflake_client.py # Snowflake connection & operations
│ └── openai_client.py # OpenAI integration
├── requirements.txt # Python dependencies
├── env.template # Environment variables template
└── README.md # This file发展
添加新工具
- 在中定义您的工具功能
src/main.py:
@mcp.tool()
async def my_new_tool(param: str, ctx: Context) -> str:
"""Description of what the tool does"""
await ctx.info(f"Processing: {param}")
# Your logic here
return "result"- 添加适当的错误处理和日志记录
- 使用FastMCP开发模式进行测试：
fastmcp dev src/main.py
添加新资源
@mcp.resource("snowflake://my-resource/{param}")
async def my_resource(param: str) -> Dict[str, Any]:
"""Resource description"""
# Your logic here
return {"data": "value"}故障排除
常见问题
- 连接错误
- 在中验证Snowflake凭据 .env - 检查网络连接 - 确保用户具有所需的权限
- OpenAI错误
- 验证 OPENAI_API_KEY 设置正确 - 检查API配额和账单 - 确保型号名称正确
- 导入错误
- 激活虚拟环境 - 安装所有要求： pip install -r requirements.txt - 从项目根目录运行
日志记录
启用调试日志记录：
LOG_LEVEL=DEBUG贡献
- 分叉存储库
- 创建要素分支
- 进行更改
- 如果适用，添加测试
- 提交拉取请求
许可证
该项目根据MIT许可证获得许可。
支持
对于问题和疑问：
- 检查故障排除部分
- 查阅FastMCP文档：https://gofastmcp.com/
- 在存储库中打开问题