Datapilot MCP Server

🚀 DataPilot MCP Server

DataPilot MCP Serverは、自然言語とAIを使用してSnowflakeと対話するための包括的なModel Context Protocol (MCP) サーバーです。FastMCP 2.0とOpenAIの統合により構築されており、AIによるガイダンスでデータを操作することができます。

✨ 主な機能

🗄️ コアデータベース操作

• execute_sql - SQLクエリを実行し、結果を取得します。
• list_databases - アクセス可能なすべてのデータベースをリストします。
• list_schemas - データベース内のスキーマをリストします。
• list_tables - データベース/スキーマ内のテーブルをリストします。
• describe_table - テーブルの詳細な列情報を取得します。
• get_table_sample - テーブルからサンプルデータを取得します。

🏭 ウェアハウス管理

• list_warehouses - 利用可能なすべてのウェアハウスをリストします。
• get_warehouse_status - 現在のウェアハウス、データベース、およびスキーマの状態を取得します。

🤖 AI機能

• natural_language_to_sql - 自然言語の質問をSQLクエリに変換します。
• analyze_query_results - AIによるクエリ結果の分析を行います。
• suggest_query_optimizations - SQLクエリの最適化提案を取得します。
• explain_query - SQLクエリを平易な英語で説明します。
• generate_table_insights - テーブルデータに関するAI生成の洞察を取得します。

📊 リソース (データアクセス)

• snowflake://databases - データベースリストにアクセスします。
• snowflake://schemas/{database} - スキーマリストにアクセスします。
• snowflake://tables/{database}/{schema} - テーブルリストにアクセスします。
• snowflake://table/{database}/{schema}/{table} - テーブルの詳細にアクセスします。

📝 プロンプト (テンプレート)

• sql_analysis_prompt - SQL分析用のテンプレートです。
• data_exploration_prompt - データ探索用のテンプレートです。
• sql_optimization_prompt - クエリ最適化用のテンプレートです。

📦 インストール

1. プロジェクトをクローンしてセットアップします:

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

2. 依存関係をインストールします:

   pip install -r requirements.txt
   

3. 環境変数を設定します:

   cp env.template .env
   # .envを編集して認証情報を入力します
   

📚 ドキュメント

環境変数

以下の設定で.envファイルを作成します。

# 必須: Snowflake接続
# アカウントの例:
# - ACCOUNT-LOCATOR.snowflakecomputing.com (推奨)
# - ACCOUNT-LOCATOR.region.cloud
# - organization-account_name
SNOWFLAKE_ACCOUNT=ACCOUNT-LOCATOR.snowflakecomputing.com
SNOWFLAKE_USER=your_username
SNOWFLAKE_PASSWORD=your_password

# オプション: デフォルトのSnowflakeコンテキスト
SNOWFLAKE_WAREHOUSE=your_warehouse_name
SNOWFLAKE_DATABASE=your_database_name
SNOWFLAKE_SCHEMA=your_schema_name
SNOWFLAKE_ROLE=your_role_name

# 必須: OpenAI API
OPENAI_API_KEY=your_openai_api_key
OPENAI_MODEL=gpt-4  # オプション、デフォルトは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

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:
        # データベースをリストする
        databases = await client.call_tool("list_databases")
        print("Databases:", databases)
        
        # 自然言語を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. 自然言語クエリ

# 自然言語で質問する
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. 実行と分析

# クエリを実行してAI分析を取得する
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. テーブルの洞察

# テーブルに関するAIによる洞察を取得する
insights = await client.call_tool("generate_table_insights", {
    "table_name": "SALES_DB.PUBLIC.CUSTOMERS",
    "sample_limit": 50
})
print(f"Table insights: {insights}")

4. クエリの最適化

# 最適化提案を取得する
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              # メインのFastMCPサーバー
│   ├── models.py            # Pydanticデータモデル
│   ├── snowflake_client.py  # Snowflake接続と操作
│   └── openai_client.py     # OpenAI統合
├── requirements.txt         # Python依存関係
├── env.template            # 環境変数のテンプレート
└── README.md              # このファイル

開発

新しいツールの追加

1. src/main.pyでツール関数を定義します:

@mcp.tool()
async def my_new_tool(param: str, ctx: Context) -> str:
    """ツールの説明"""
    await ctx.info(f"Processing: {param}")
    # ここにロジックを記述
    return "result"

2. 適切なエラーハンドリングとロギングを追加します
3. FastMCPの開発モードでテストします: fastmcp dev src/main.py

新しいリソースの追加

@mcp.resource("snowflake://my-resource/{param}")
async def my_resource(param: str) -> Dict[str, Any]:
    """リソースの説明"""
    # ここにロジックを記述
    return {"data": "value"}

トラブルシューティング

一般的な問題

1. 接続エラー

   • .env内のSnowflake認証情報を確認します
   • ネットワーク接続を確認します
   • ユーザーに必要な権限があることを確認します

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/
• リポジトリに問題を登録してください