MCP SQLite サーバー（読み取り専用）

本番環境対応の Model Context Protocol サーバーで、AI エージェントに SQLite データベース（shop.db）への安全な読み取り専用アクセスを提供します。公式の mcp Python SDK を stdio トランスポートで使用して構築されています。

特徴

• 3 つの MCP ツール: list_tables、describe_table、query_database

• 多層防御による読み取り専用の安全性: SQLite URI 読み取り専用モード + PRAGMA query_only + SQL バリデーター + EXPLAIN オペコード検査

• クエリ検証: INSERT/UPDATE/DELETE/DROP/ALTER/CREATE/REPLACE/TRUNCATE/ATTACH/DETACH、複文クエリ（;）、SQL コメント（--、/* */）、変更を伴う PRAGMA を拒否 — 文字列リテラルによる誤検出はありません

• ページネーション: デフォルトの行数制限（100）、limit/offset パラメータ、切り詰め出力フラグ

• stderr のみのログ: すべてのログとトレースバックは sys.stderr に出力されます。stdout は JSON-RPC 専用です

• 完全な型ヒント: mypy --strict でクリーン

• TDD: セキュリティ、DB レイヤー、MCP ツール、8 つのベンチマーククエリ、stderr ガードをカバーする 105 のテスト

Related MCP server: sqlite-mcp-server

クイックスタート

前提条件

• Python 3.10+

• SQLite データベースファイル（デフォルト: ./shop.db）

ローカルセットアップ

python -m venv .venv
source .venv/bin/activate
pip install -e ".[dev]"

設定

.env.example をコピーし、データベースパスを設定します:

cp .env.example .env
# Edit DATABASE_PATH to point to your SQLite file

または、環境変数を直接設定します:

export DATABASE_PATH=/abs/path/to/shop.db

サーバーの実行

python -m mcp_server.server

サーバーは MCP stdio トランスポートを使用して stdin/stdout で通信します。直接操作する必要はありません — MCP クライアント（例: Claude Desktop、AI エージェント）が接続します。

MCP クライアント設定

標準 Python

MCP クライアント設定（例: Claude Desktop の claude_desktop_config.json）に次を追加します:

{
  "mcpServers": {
    "sqlite-shop": {
      "command": "python",
      "args": ["-m", "mcp_server.server"],
      "env": {
        "DATABASE_PATH": "/abs/path/to/shop.db"
      }
    }
  }
}

Docker

まずイメージをビルドします:

docker build -t mcp-shop:latest .

次に MCP クライアントを設定します:

{
  "mcpServers": {
    "sqlite-shop": {
      "command": "docker",
      "args": [
        "run", "-i", "--rm",
        "-v", "/abs/path/to/shop.db:/app/shop.db",
        "-e", "DATABASE_PATH=/app/shop.db",
        "mcp-shop:latest"
      ]
    }
  }
}

Docker Compose

docker compose up -d

ツール

list_tables

データベース内のすべてのユーザーテーブルとビューを一覧表示します（内部の sqlite_* テーブルは除外）。

パラメータ: なし

戻り値:

{
  "tables": ["customers", "orders", "order_items", "products"],
  "count": 4
}

describe_table

テーブルのスキーマ（列、外部キー、行数、CREATE 文）を説明します。

パラメータ:

• table（文字列、必須）: 説明するテーブルの名前。

戻り値:

{
  "table": "customers",
  "columns": [
    {"cid": 0, "name": "id", "type": "INTEGER", "notnull": 0, "default": null, "pk": 1},
    {"cid": 1, "name": "first_name", "type": "TEXT", "notnull": 1, "default": null, "pk": 0}
  ],
  "foreign_keys": [],
  "row_count": 150,
  "sql": "CREATE TABLE customers (...)"
}

query_database

ページネーションをサポートする読み取り専用 SQL クエリを実行します。

パラメータ:

• sql（文字列、必須）: 単一の読み取り専用 SQL 文（SELECT、WITH、EXPLAIN、または読み取り専用 PRAGMA）。

• limit（整数、任意）: 返す最大行数。デフォルト: 100。最大: 1000。

• offset（整数、任意）: スキップする行数。デフォルト: 0。

戻り値:

{
  "columns": ["id", "first_name"],
  "rows": [{"id": 1, "first_name": "Alice"}, {"id": 2, "first_name": "Bob"}],
  "row_count": 2,
  "truncated": false,
  "limit": 100,
  "offset": 0
}

truncated が true の場合、さらに行が利用可能です — offset を増やして次のページを取得します。

セキュリティ

サーバーは読み取り専用アクセスを保証するために多層防御を実装しています:

レイヤー 1: SQLite 接続（URI 読み取り専用モード）

データベースは file:<path>?mode=ro で開かれ、SQLite エンジンレベルでの書き込みを防ぎます。さらに、すべての接続で PRAGMA query_only = ON が設定されます。

レイヤー 2: SQL クエリバリデーター（security.py）

クエリが SQLite に到達する前に、多段階のバリデーターを通過します:

1. 文字列リテラルの除去: 文字列リテラル（'...'、"..."）はプレースホルダーに置き換えられ、データ内のキーワード（例: "Deleted Item" という製品名）が誤検出を引き起こさないようにします。

2. コメント検出: SQL コメント（--、/* */）は拒否され、コメントベースのバイパスを防ぎます。

3. 複文の拒否: セミコロン（;）は拒否され、スタッククエリを防ぎます。

4. キーワード分析: 最初の実ステートメントキーワードは SELECT、WITH、EXPLAIN、または PRAGMA でなければなりません。破壊的なキーワード（INSERT、UPDATE、DELETE、DROP、ALTER、CREATE、REPLACE、TRUNCATE、ATTACH、DETACH、VACUUM など）はブロックされます。

5. PRAGMA 検証: 読み取り専用の PRAGMA（table_info、database_list など）は許可されます。代入（=）を含む PRAGMA や、変更を伴う PRAGMA のブロックリスト（journal_mode、synchronous、foreign_keys など）に含まれる PRAGMA は拒否されます。

レイヤー 3: EXPLAIN オペコード検査

最終防御として、クエリは EXPLAIN <query> を介して SQLite 自身のパーサーで実行されます。結果のオペコードストリームは、書き込みオペコード（OpenWrite、Insert、Delete、Create、Drop など）と書き込みトランザクションフラグについて検査されます。見つかった場合、クエリは拒否されます。

レイヤー 4: サニタイズされたエラーメッセージ

クライアントに返されるすべてのエラーはサニタイズされ、ファイルシステムパスや内部詳細は情報漏洩を防ぐために除去されます。

テスト

テストは一時データベースまたはインメモリデータベースのみを使用します — 本番の shop.db は使用しません。

# Run all tests
python -m pytest

# Run with verbose output
python -m pytest -v

# Run a specific test file
python -m pytest tests/test_security.py

テストカバレッジ

静的解析

# Type checking
python -m mypy

# Linting
python -m ruff check src/ tests/

プロジェクト構造

.
├── .env.example          # Environment variable template
├── Dockerfile            # Docker containerization
├── docker-compose.yml    # Docker Compose config
├── pyproject.toml        # Package config, deps, tool settings
├── README.md             # This file
├── shop.db               # The SQLite database (not included in tests)
├── src/mcp_server/
│   ├── __init__.py
│   ├── config.py         # Configuration (DATABASE_PATH, limits, URI builder)
│   ├── db.py             # Read-only Database class with introspection + query
│   ├── security.py       # SQL validator (multi-layer defense-in-depth)
│   ├── server.py         # MCP server entrypoint (stdio transport)
│   ├── tools.py          # MCP tool definitions and handlers
│   └── py.typed          # PEP 561 marker
└── tests/
    ├── __init__.py
    ├── test_db.py        # Database layer + benchmark tests
    ├── test_security.py  # Query validator tests
    └── test_server.py    # MCP server/tool tests

ベンチマークタスク

サーバーのツールにより、AI エージェントは次の分析タスクを実行できます（制御されたフィクスチャデータベースに対するテストで検証済み）:

1. テーブル発見: list_tables + describe_table — すべてのテーブルを一覧表示し、スキーマを説明します。

2. フィルタリングされたカウント: SELECT COUNT(*) FROM customers WHERE country = 'Germany' を使用した query_database。

3. 国別集計: SELECT country, COUNT(*) ... GROUP BY country ORDER BY ... DESC LIMIT 1。

4. 顧客 LTV: customers + orders を結合し、SUM(total_amount)、合計で並べ替え。

5. 製品パフォーマンス: order_items + products を結合し、数量と売上で集計、LIMIT 5。

6. カテゴリ集計: order_items → products → category を辿り、売上を集計、LIMIT 3。

7. 日付フィルタリング: SUM(total_amount) WHERE substr(order_date,1,4) = '2025'。

8. 注文集計: customers + orders を結合し、COUNT(o.id)、件数で並べ替え。

設定

ライセンス

このプロジェクトはデモンストレーション目的で現状のまま提供されます。