Data Pilot (Snowflake)

Um servidor abrangente do Model Context Protocol (MCP) para interagir com o Snowflake usando linguagem natural e IA.

GitHub
2
Experimente este MCPPatrocinado

Documentação

Servidor MCP DataPilot

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

Navegue pelos seus dados com orientação de IA. Um servidor abrangente de Model Context Protocol (MCP) para interagir com o Snowflake usando linguagem natural e IA. Construído com FastMCP 2.0 e integração com OpenAI.

Recursos

🗄️ Operações Principais de Banco de Dados

  • execute_sql - Executar consultas SQL com resultados
  • list_databases - Listar todos os bancos de dados acessíveis
  • list_schemas - Listar esquemas em um banco de dados
  • list_tables - Listar tabelas em um banco de dados/esquema
  • describe_table - Obter informações detalhadas das colunas de uma tabela
  • get_table_sample - Recuperar dados de amostra de tabelas

🏭 Gerenciamento de Warehouse

  • list_warehouses - Listar todos os warehouses disponíveis
  • get_warehouse_status - Obter o status atual do warehouse, banco de dados e esquema

🤖 Recursos com IA

  • natural_language_to_sql - Converter perguntas em linguagem natural para consultas SQL
  • analyze_query_results - Análise dos resultados de consultas com IA
  • suggest_query_optimizations - Obter sugestões de otimização para consultas SQL
  • explain_query - Explicações em inglês simples de consultas SQL
  • generate_table_insights - Insights gerados por IA sobre dados de tabelas

📊 Recursos (Acesso a Dados)

  • snowflake://databases - Acessar lista de bancos de dados
  • snowflake://schemas/{database} - Acessar lista de esquemas
  • snowflake://tables/{database}/{schema} - Acessar lista de tabelas
  • snowflake://table/{database}/{schema}/{table} - Acessar detalhes de tabelas

📝 Prompts (Modelos)

  • sql_analysis_prompt - Modelos para análise SQL
  • data_exploration_prompt - Modelos para exploração de dados
  • sql_optimization_prompt - Modelos para otimização de consultas

Instalação

  1. Clone e configure o projeto:

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

  2. Instale as dependências:

    pip install -r requirements.txt

  3. Configure as variáveis de ambiente:

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

Configuração

Variáveis de Ambiente

Crie um arquivo .env com a seguinte configuração:

# 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

Configuração da Conta Snowflake

  1. Obtenha o identificador da sua conta Snowflake - Vários formatos suportados:

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

  2. Garanta que seu usuário tenha as permissões adequadas:

    • USAGE em warehouses, bancos de dados e esquemas
    • SELECT em tabelas para consultas
    • SHOW privilégios para listar objetos

Uso

Executando o Servidor

Método 1: Execução direta

python -m src.main

Método 2: Usando a CLI do FastMCP

fastmcp run src/main.py

Método 3: Modo de desenvolvimento com recarga automática

fastmcp dev src/main.py

Conectando a Clientes MCP

Claude Desktop

Adicione à configuração do 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 o 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)

Exemplo de Uso

1. Consulta em Linguagem 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. Executar e Analisar

# 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. Insights de Tabelas

# 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. Otimização 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}")

Arquitetura

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

Estrutura do Projeto

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

Desenvolvimento

Adicionando Novas Ferramentas

  1. Defina sua função de ferramenta em 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. Adicione tratamento de erros e registro de logs adequados
  2. Teste com o modo de desenvolvimento do FastMCP: fastmcp dev src/main.py

Adicionando Novos Recursos

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

Solução de Problemas

Problemas Comuns

  1. Erros de Conexão

    • Verifique as credenciais do Snowflake em .env
    • Verifique a conectividade de rede
    • Garanta que o usuário tenha as permissões necessárias

  2. Erros do OpenAI

    • Verifique se OPENAI_API_KEY está configurado corretamente
    • Verifique a cota e o faturamento da API
    • Garanta que o nome do modelo esteja correto

  3. Erros de Importação

    • Ative o ambiente virtual
    • Instale todos os requisitos: pip install -r requirements.txt
    • Execute a partir do diretório raiz do projeto

Registro de Logs

Ative o registro de depuração:

LOG_LEVEL=DEBUG

Contribuição

  1. Faça um fork do repositório
  2. Crie um branch de funcionalidade
  3. Faça suas alterações
  4. Adicione testes, se aplicável
  5. Envie um pull request

Licença

Este projeto está licenciado sob a Licença MIT.

Suporte

Para problemas e dúvidas:

  • Consulte a seção de solução de problemas
  • Revise a documentação do FastMCP: https://gofastmcp.com/
  • Abra um problema no repositório