Data Pilot (Snowflake)

A comprehensive Model Context Protocol (MCP) server for interacting with Snowflake using natural language and AI.

Navigate your data with AI guidance. A comprehensive Model Context Protocol (MCP) server for interacting with Snowflake using natural language and AI. Built with FastMCP 2.0 and OpenAI integration.

- execute_sql- Execute SQL queries with results
- list_databases- List all accessible databases
- list_schemas- List schemas in a database
- list_tables- List tables in a database/schema
- describe_table- Get detailed table column information
- get_table_sample- Retrieve sample data from tables

- list_warehouses- List all available warehouses
- get_warehouse_status- Get current warehouse, database, and schema status

- natural_language_to_sql- Convert natural language questions to SQL queries
- analyze_query_results- AI-powered analysis of query results
- suggest_query_optimizations- Get optimization suggestions for SQL queries
- explain_query- Plain English explanations of SQL queries
- generate_table_insights- AI-generated insights about table data

- snowflake://databases- Access database list
- snowflake://schemas/{database}- Access schema list
- snowflake://tables/{database}/{schema}- Access table list
- snowflake://table/{database}/{schema}/{table}- Access table details

- sql_analysis_prompt- Templates for SQL analysis
- data_exploration_prompt- Templates for data exploration
- sql_optimization_prompt- Templates for query optimization

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

cp env.template .env # Edit .env with your credentials

Create a.envfile with the following configuration:

# 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

-

Get your Snowflake account identifier- Multiple formats supported:

- Recommended:ACCOUNT-LOCATOR.snowflakecomputing.com(e.g.,SCGEENJ-UR66679.snowflakecomputing.com)
- Regional:ACCOUNT-LOCATOR.region.cloud(e.g.,xy12345.us-east-1.aws)
- Legacy:organization-account_name

Ensure your user has appropriate permissions:

- USAGEon warehouses, databases, and schemas
- SELECTon tables for querying
- SHOWprivileges for listing objects

Method 3: Development mode with auto-reload

Add to your Claude Desktop configuration:

{ "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" } } } }

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)

# 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}")

# 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}")

# 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}")

# 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

- Define your tool function insrc/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"

- Add appropriate error handling and logging
- Test with FastMCP dev mode: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"}

- Verify Snowflake credentials in.env
- Check network connectivity
- Ensure user has required permissions

- VerifyOPENAI_API_KEYis set correctly
- Check API quota and billing
- Ensure model name is correct

- Activate virtual environment
- Install all requirements:pip install -r requirements.txt
- Run from project root directory
- Fork the repository
- Create a feature branch
- Make your changes
- Add tests if applicable
- Submit a pull request

This project is licensed under the MIT License.

- Check the troubleshooting section
- Review FastMCP documentation:https://gofastmcp.com/
- Open an issue in the repository

A Python-based MCP server that connects to an Elasticsearch index with Apple HealthKit data.