Data Pilot (Snowflake)

Un servidor completo del Protocolo de Contexto de Modelo (MCP) para interactuar con Snowflake usando lenguaje natural e IA.

GitHub
2
Prueba este MCPPatrocinado

Documentación

Servidor MCP de DataPilot

CI/CD Pipeline Coverage Status Python Version License: MIT Code style: black Security: bandit Pre-commit

Navega por tus datos con asistencia de IA. Un servidor integral de Protocolo de Contexto de Modelo (MCP) para interactuar con Snowflake usando lenguaje natural e IA. Construido con FastMCP 2.0 e integración con OpenAI.

Características

🗄️ Operaciones Principales de Base de Datos

  • execute_sql - Ejecuta consultas SQL con resultados
  • list_databases - Lista todas las bases de datos accesibles
  • list_schemas - Lista los esquemas en una base de datos
  • list_tables - Lista las tablas en una base de datos/esquema
  • describe_table - Obtén información detallada de las columnas de una tabla
  • get_table_sample - Recupera datos de muestra de las tablas

🏭 Gestión de Almacenes

  • list_warehouses - Lista todos los almacenes disponibles
  • get_warehouse_status - Obtén el estado actual del almacén, la base de datos y el esquema

🤖 Funciones Impulsadas por IA

  • natural_language_to_sql - Convierte preguntas en lenguaje natural a consultas SQL
  • analyze_query_results - Análisis de resultados de consultas impulsado por IA
  • suggest_query_optimizations - Obtén sugerencias de optimización para consultas SQL
  • explain_query - Explicaciones en inglés sencillo de consultas SQL
  • generate_table_insights - Información generada por IA sobre los datos de las tablas

📊 Recursos (Acceso a Datos)

  • snowflake://databases - Accede a la lista de bases de datos
  • snowflake://schemas/{database} - Accede a la lista de esquemas
  • snowflake://tables/{database}/{schema} - Accede a la lista de tablas
  • snowflake://table/{database}/{schema}/{table} - Accede a los detalles de las tablas

📝 Prompts (Plantillas)

  • sql_analysis_prompt - Plantillas para análisis SQL
  • data_exploration_prompt - Plantillas para exploración de datos
  • sql_optimization_prompt - Plantillas para optimización de consultas

Instalación

  1. Clona y configura el proyecto:

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

  2. Instala las dependencias:

    pip install -r requirements.txt

  3. Configura las variables de entorno:

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

Configuración

Variables de Entorno

Crea un archivo .env con la siguiente configuración:

# 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

Configuración de la Cuenta de Snowflake

  1. Obtén tu identificador de cuenta de Snowflake - Se admiten múltiples formatos:

    • Recomendado: ACCOUNT-LOCATOR.snowflakecomputing.com (por ejemplo, SCGEENJ-UR66679.snowflakecomputing.com)
    • Regional: ACCOUNT-LOCATOR.region.cloud (por ejemplo, xy12345.us-east-1.aws)
    • Legado: organization-account_name

  2. Asegúrate de que tu usuario tenga los permisos adecuados:

    • USAGE en almacenes, bases de datos y esquemas
    • SELECT en tablas para consultas
    • SHOW privilegios para listar objetos

Uso

Ejecutando el Servidor

Método 1: Ejecución directa

python -m src.main

Método 2: Usando la CLI de FastMCP

fastmcp run src/main.py

Método 3: Modo de desarrollo con recarga automática

fastmcp dev src/main.py

Conectándose a Clientes MCP

Claude Desktop

Agrega a tu configuración de 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"
      }
    }
  }
}

Usando el Cliente 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)

Ejemplo de Uso

1. Consulta en Lenguaje Natural

# 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. Ejecutar y Analizar

# 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. Información de Tablas

# 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. Optimización de Consultas

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

Arquitectura

┌─────────────────┐    ┌─────────────────┐    ┌─────────────────┐
│   MCP Client    │    │   FastMCP       │    │   Snowflake     │
│   (Claude/etc)  │◄──►│   Server        │◄──►│   Database      │
└─────────────────┘    └─────────────────┘    └─────────────────┘
                                │
                                ▼
                       ┌─────────────────┐
                       │   OpenAI API    │
                       │   (GPT-4)       │
                       └─────────────────┘

Estructura del Proyecto

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

Desarrollo

Agregando Nuevas Herramientas

  1. Define tu función de herramienta en 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. Agrega manejo de errores y registro adecuados
  2. Prueba con el modo de desarrollo de FastMCP: fastmcp dev src/main.py

Agregando Nuevos Recursos

@mcp.resource("snowflake://my-resource/{param}")
async def my_resource(param: str) -> Dict[str, Any]:
    """Resource description"""
    # Your logic here
    return {"data": "value"}

Solución de Problemas

Problemas Comunes

  1. Errores de Conexión

    • Verifica las credenciales de Snowflake en .env
    • Comprueba la conectividad de red
    • Asegúrate de que el usuario tenga los permisos requeridos

  2. Errores de OpenAI

    • Verifica que OPENAI_API_KEY esté configurado correctamente
    • Comprueba la cuota y facturación de la API
    • Asegúrate de que el nombre del modelo sea correcto

  3. Errores de Importación

    • Activa el entorno virtual
    • Instala todos los requisitos: pip install -r requirements.txt
    • Ejecuta desde el directorio raíz del proyecto

Registro

Habilita el registro de depuración:

LOG_LEVEL=DEBUG

Contribuciones

  1. Haz un fork del repositorio
  2. Crea una rama de características
  3. Realiza tus cambios
  4. Agrega pruebas si corresponde
  5. Envía una solicitud de extracción

Licencia

Este proyecto está licenciado bajo la Licencia MIT.

Soporte

Para problemas y preguntas:

  • Consulta la sección de solución de problemas
  • Revisa la documentación de FastMCP: https://gofastmcp.com/
  • Abre un problema en el repositorio