はじめに

最近、データ活用の相談が増えています。ただ、事業部のメンバーが自分でデータを取り出すのは、まだハードルが高いのが実情です。
そこで、「先月のチャネル別の売上は？」と、事業部のメンバーが自然な言葉でデータに聞けるようにしたいと考えました。MCP（Model Context Protocol）を使えば、Claude からデータベースにつなぐこと自体は難しくありません。

難しいのは、業務で使ってよいと言えるくらい安全にすることです。

AIが誤ってデータを書き換えたり消したりしないか
お客様の氏名・メールアドレス・電話番号がAIに渡らないか
重いクエリでデータベースを止めないか
誰が何を聞いたか、あとから追えるか

これらに対策した、ECの受注データ用の読み取り専用MCPサーバーを作って公開しました。データはすべて手元で生成する架空のものです。

Claude からできること

ツール	用途
`describe_schema`	テーブル・列と、売上や日付の定義
`search_orders`	ステータス・チャネル・期間・顧客で注文を絞り込む
`get_order`	1件の注文を明細とマスキングした顧客情報つきで返す
`sales_summary`	日・週・月・チャネル・カテゴリ別の売上・件数・平均単価
`top_products`	売上または数量の多い商品
`run_sql`	上のツールで答えられない質問のための、ガード付きの `SELECT` 1文

ほかに、スキーマを返すリソース（schema://tables）と、月次レポートを作るプロンプト（monthly_report）も用意しています。

多層防御の全体像

どの層も「前の層が破られるかもしれない」前提で作っています。

層	防ぐもの
DBを `mode=ro` で開く	ほかのすべてのチェックを抜けられても、書き込みはできない
固定のパラメータ化ツールを優先	よくある質問でのSQLインジェクションや、際限のないクエリ
`run_sql` は `SELECT` / `WITH` の1文だけ	`SELECT 1; DROP ...` のような複文
SQLite の authorizer	書き込み・`PRAGMA`・`ATTACH`・`load_extension`・内部テーブル・個人情報の列の読み取り
時間と件数の上限	無限に回る再帰CTEや、大量の結果
`get_order` のマスキング	個人情報がサーバーの外に出ること
ツールのアノテーション	クライアントに全ツールが読み取り専用（`readOnlyHint: true`）だと伝える
監査ログ（JSONL）	誰が何を聞き、何件返したか追えない状態。結果そのものは残さない

1. 最後の砦：読み取り専用で開く

ec_orders_mcp/store.py（抜粋）

con = sqlite3.connect(f"file:{self.db_path}?mode=ro", uri=True)

たった1行ですが、これがあれば、上の層をすべてすり抜けるクエリが来ても書き込みはできません。

2. SQLite の authorizer で「読んでいい列」まで制御する

run_sql を使う接続には、SQLite の認可コールバック（set_authorizer）を設定しています。SQLの解析時に、操作の種類やテーブル・列ごとに呼ばれるので、文字列のチェックよりも確実です。

ec_orders_mcp/store.py（抜粋）

PII_COLUMNS = {("customers", "email"), ("customers", "phone"), ("customers", "name")}

_ALLOWED_ACTIONS = {sqlite3.SQLITE_SELECT, sqlite3.SQLITE_READ, sqlite3.SQLITE_FUNCTION, sqlite3.SQLITE_RECURSIVE}
_DENIED_FUNCTIONS = {"load_extension", "readfile", "writefile", "edit"}


def _authorizer(action, arg1, arg2, db_name, trigger):
    if action not in _ALLOWED_ACTIONS:
        return sqlite3.SQLITE_DENY
    if action == sqlite3.SQLITE_READ:
        table, column = (arg1 or "").lower(), (arg2 or "").lower()
        if table.startswith("sqlite_") or (table, column) in PII_COLUMNS:
            return sqlite3.SQLITE_DENY
    if action == sqlite3.SQLITE_FUNCTION and (arg2 or "").lower() in _DENIED_FUNCTIONS:
        return sqlite3.SQLITE_DENY
    return sqlite3.SQLITE_OK

「許可したテーブル以外は拒否」という書き方にすると、WITH で定義したCTEの名前まで拒否されてしまいます。そこで、内部テーブル（sqlite_*）と個人情報の列だけを拒否する形にしています（DBファイルは読み取り専用で、ほかに何も入っていない前提）。再帰CTEを使えるようにするには、SQLITE_RECURSIVE の許可も必要です。

3. 暴走するクエリを止める

ec_orders_mcp/store.py（抜粋）

deadline = time.monotonic() + TIME_BUDGET_SECONDS  # 2秒
con.set_progress_handler(lambda: 1 if time.monotonic() > deadline else 0, 10_000)

set_progress_handler は、一定数の命令ごとに呼ばれ、0以外を返すとクエリを中断します。終わらない再帰CTEもこれで止まります。件数は、元のクエリを SELECT * FROM (...) LIMIT 201 で包んで上限をかけ、200件を超えたら「切り詰めた」と返します。

4. エラーは「AIが次に直せる」言葉で返す

拒否したときは、何がだめで、なぜなのかをツールエラーとして返します。

ec_orders_mcp/store.py（抜粋）

raise QueryRejected(
    "Query touches a restricted table, column or function. "
    "Personal data (customers.name, email, phone) is not available through SQL."
)

理由がわかれば、Claude は同じクエリをやみくもに再試行せず、聞き方を変えてくれます。

5. 個人情報はマスキングして返す

get_order は顧客情報を返しますが、マスキングしています。

ec_orders_mcp/store.py（抜粋）

def mask_email(email: str) -> str:
    """customer00012@example.com -> c***2@example.com"""
    local, _, domain = email.partition("@")
    if not domain:
        return "***"
    return f"{local[0]}***{local[-1]}@{domain}" if len(local) > 1 else f"*@{domain}"


def mask_phone(phone: str) -> str:
    """090-1234-5678 -> ***-****-5678"""
    digits = re.sub(r"\D", "", phone)
    return f"***-****-{digits[-4:]}" if len(digits) >= 4 else "***"

6. 固定のツールを先に、SQLは最後の手段に

サーバーの説明（instructions）で、Claude に使い方を伝えています。

ec_orders_mcp/server.py（抜粋）

INSTRUCTIONS = """\
Read-only access to an online shop's order data (synthetic).
Start with describe_schema if you need table or column details.
Prefer the fixed tools (search_orders, get_order, sales_summary, top_products);
use run_sql only for questions they cannot answer. Personal data is masked or
unavailable by design: do not try to work around it.
Dates are ISO strings (YYYY-MM-DD). Revenue counts completed orders only.
"""

業務の質問の多くは、名前のわかりやすい少数のツールで答えられます。自由なSQLより速く、トークンも少なく、監査もしやすいので、SQLはガード付きの予備にしています。「売上は完了した注文だけを数える」という定義も、ここと describe_schema に書いておき、AIとテストが同じ定義を使うようにしています。

Claude から使う

Python 3.11 以上と、MCP Python SDK（v2）を使っています。

git clone https://github.com/yahayuta/ec-orders-mcp.git
cd ec-orders-mcp
pip install -e .
claude mcp add ec-orders -e EC_MCP_DB="$PWD/data/orders.db" -- python -m ec_orders_mcp

Claude Desktop なら claude_desktop_config.json に登録します（READMEに例があります）。DBファイルがなければ、起動時に架空のデータで自動生成されます。

テスト

テストは23件です。

ガード：書き込み・複文・PRAGMA・ATTACH・拡張・内部テーブル・個人情報の列がすべて拒否されること
マスキング、件数の上限、時間の上限、読み取り専用の接続、監査ログ
固定ツールの集計と、同じ内容の生SQLの結果が一致すること
サーバー本体を、プロセス内と、Claude Desktop と同じ stdio 起動の両方で呼び出すこと

CIでは Python 3.11〜3.13 で実行しています。

まとめ

「つなぐ」より「安全にする」ほうが本題
層ごとに「前の層が破られる」前提で守る（読み取り専用・authorizer・上限・マスキング・監査）
エラーメッセージはAIが次の一手を直せるように書く
固定のツールを先に、自由なSQLは最後の手段に
裏側は薄いクラスなので、BigQuery やレプリカに差し替えても、ツールの約束はそのまま使える

私の職場でもここ数年でBigQueryの活用が進んでいるので、裏側をBigQueryに差し替えれば、事業部のメンバーが安全にデータに聞ける仕組みとして使えると考えています。

リポジトリ：https://github.com/yahayuta/ec-orders-mcp