AI导哈

AI数据库交互服务器

一个基于FastMCP 2.0和OpenAI集成的Model Context Protocol (MCP)服务器，用于通过自然语言和AI与Snowflake数据库进行交互。

  • by rickyb30
  • Python
  • 本地部署
  • 2 星标
  • 更新于
在 GitHub 查看 rickyb30/datapilot-mcp-server

使用说明

数据助理MCP服务器

在人工智能指导下导航您的数据。一个全面的模型上下文协议（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 -查询优化模板

安装

  1. 克隆并设置项目：

    git clone <repository-url>
cd datapilot
python -m venv venv
source venv/bin/activate  # On Windows: venv\Scripts\activate

  2. 安装依赖项：

    pip install -r requirements.txt

  3. 配置环境变量：

    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-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 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

发展

添加新工具

  1. 在中定义您的工具功能 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"
  1. 添加适当的错误处理和日志记录
  2. 使用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"}

故障排除

常见问题

  1. 连接错误

    • 在中验证Snowflake凭据 .env
    • 检查网络连接
    • 确保用户具有所需的权限

  2. OpenAI错误

    • 验证 OPENAI_API_KEY 设置正确
    • 检查API配额和账单
    • 确保型号名称正确

  3. 导入错误

    • 激活虚拟环境
    • 安装所有要求： pip install -r requirements.txt
    • 从项目根目录运行

日志记录

启用调试日志记录：

LOG_LEVEL=DEBUG

贡献

  1. 分叉存储库
  2. 创建要素分支
  3. 进行更改
  4. 如果适用，添加测试
  5. 提交拉取请求

许可证

该项目根据MIT许可证获得许可。

支持

对于问题和疑问：

  • 检查故障排除部分
  • 查阅FastMCP文档：https://gofastmcp.com/
  • 在存储库中打开问题

相关 MCP