10
13

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?

業務DBを読み取り専用MCPにしてClaudeから参照する

10
Posted at

はじめに

AIから社内データを参照できるようにすると、「未対応のタスクを確認して」「この顧客の状態を調べて」といった自然文から業務データを扱えるようになります。

一方、業務DBへAIをつなぐときに気になるのが権限です。便利だからといって、SQLを自由に実行できる接続をそのまま渡すのは避けたいところです。

私はAIを使った書類の読み取りやデータ整理などに取り組むとき、まず読み取り専用から始めることを大切にしています。

今回はその考え方をMCP(Model Context Protocol)に当てはめ、Claudeから参照できる小さなMCPサーバーを作ります。持ち帰れるものは、動かせるサンプルコードと、実際の業務DBへつなぐ前のチェックリストです。

結論

ClaudeへSQL実行権限を渡さず、MCP側で許可した読み取り操作だけを公開します。

さらにDB接続そのものも読み取り専用にし、「AIが書き込まない」ではなく「仕組みとして書き込めない」構成にします。

環境

この記事では次の環境を前提にします。

OS: macOS / Linux
Python: 3.12
MCP Python SDK: 2.2.0
Database: SQLite 3
Client: Claude Code

MCP Python SDKはPython向けの公式SDKです。実際に試す際は、記事末尾の公式リポジトリで現在のバージョンを確認してください。

今回は構成を小さくするためSQLiteを使います。本番の業務DBがPostgreSQLやMySQLなどでも、「公開する操作を限定する」という考え方は同じです。

全体像は次のようになります。

Claudeから利用できるのは、

get_customer
search_customers
list_open_tasks

という3つの操作です。

execute_sqlのような万能ツールは作りません。

実装

1. プロジェクトを準備する

作業用ディレクトリと仮想環境を作ります。

mkdir readonly-business-mcp
cd readonly-business-mcp

python -m venv .venv
source .venv/bin/activate

MCP Python SDKをインストールします。

pip install "mcp[cli]==2.2.0"

ファイルは次の3つだけです。

readonly-business-mcp/
├── server.py
├── create_sample_db.py
└── business.db

2. サンプルDBを作る

ここから登場する会社名やデータはすべて架空です。

顧客と対応タスクを管理している、小さな業務DBを想定します。

# create_sample_db.py

import sqlite3

DB_PATH = "business.db"

connection = sqlite3.connect(DB_PATH)

connection.executescript(
    """
    DROP TABLE IF EXISTS tasks;
    DROP TABLE IF EXISTS customers;

    CREATE TABLE customers (
        id INTEGER PRIMARY KEY,
        name TEXT NOT NULL,
        status TEXT NOT NULL
    );

    CREATE TABLE tasks (
        id INTEGER PRIMARY KEY,
        customer_id INTEGER NOT NULL,
        title TEXT NOT NULL,
        status TEXT NOT NULL,
        FOREIGN KEY (customer_id) REFERENCES customers(id)
    );

    INSERT INTO customers (id, name, status) VALUES
        (1, 'サンプル商事', 'active'),
        (2, 'テスト工業', 'active'),
        (3, '例示サービス', 'inactive');

    INSERT INTO tasks (id, customer_id, title, status) VALUES
        (1, 1, '契約内容を確認', 'open'),
        (2, 1, '次回打ち合わせ準備', 'done'),
        (3, 2, '問い合わせ内容を確認', 'open');
    """
)

connection.commit()
connection.close()

実行します。

python create_sample_db.py

これでbusiness.dbができます。

3. DB接続を読み取り専用にする

MCPのtoolに更新処理を書かなければ十分、とは考えません。

将来コードを変更したとき、誤って更新SQLを追加する可能性があります。そのためSQLiteへの接続自体を読み取り専用にします。

from pathlib import Path
import sqlite3

DB_PATH = Path(__file__).with_name("business.db").resolve()


def get_connection() -> sqlite3.Connection:
    connection = sqlite3.connect(
        f"file:{DB_PATH}?mode=ro",
        uri=True,
    )
    connection.row_factory = sqlite3.Row
    return connection

SQLiteのURI filenameでmode=roを指定しています。

考え方としては、

MCP tool
  ↓
SELECTだけ実装

DB connection
  ↓
read-only

という二段構えです。

PostgreSQLなどへ置き換える場合は、MCP専用のDBユーザーを作り、必要なテーブルやビューへのSELECTだけを許可する構成を検討します。

4. なぜ自由なSQLを公開しないのか

次のようなtoolを作れば、かなり自由な問い合わせができます。

# 今回は採用しない

@mcp.tool()
def execute_sql(sql: str):
    with get_connection() as connection:
        return connection.execute(sql).fetchall()

ClaudeにSQLを作らせれば、質問に合わせて柔軟に検索できます。

ただし、この方式ではMCPサーバーが任意SQLの実行装置になります。

読み取り専用接続ならUPDATEやDELETEは防げても、

SELECT * FROM customers;

のような広すぎる取得は可能です。

つまり、

書き込みを防ぐ

ことと、

必要な情報だけ読ませる

ことは同じではありません。

そこで今回は、SQLではなく「顧客を検索する」「未完了タスクを取得する」という業務操作を公開します。

5. MCPサーバーを作る

server.pyを作ります。

from pathlib import Path
import sqlite3
from typing import Any

from mcp.server import MCPServer

DB_PATH = Path(__file__).with_name("business.db").resolve()

mcp = MCPServer("readonly-business-db")


def get_connection() -> sqlite3.Connection:
    connection = sqlite3.connect(
        f"file:{DB_PATH}?mode=ro",
        uri=True,
    )
    connection.row_factory = sqlite3.Row
    return connection


def row_to_dict(row: sqlite3.Row) -> dict[str, Any]:
    return dict(row)


@mcp.tool()
def get_customer(customer_id: int) -> dict[str, Any]:
    """顧客IDから基本情報を取得します。"""

    with get_connection() as connection:
        row = connection.execute(
            """
            SELECT id, name, status
            FROM customers
            WHERE id = ?
            """,
            (customer_id,),
        ).fetchone()

    if row is None:
        return {"found": False, "customer": None}

    return {
        "found": True,
        "customer": row_to_dict(row),
    }


@mcp.tool()
def search_customers(
    keyword: str,
    limit: int = 10,
) -> dict[str, Any]:
    """顧客名を部分一致で検索します。"""

    safe_limit = max(1, min(limit, 20))

    with get_connection() as connection:
        rows = connection.execute(
            """
            SELECT id, name, status
            FROM customers
            WHERE name LIKE ?
            ORDER BY id
            LIMIT ?
            """,
            (f"%{keyword}%", safe_limit),
        ).fetchall()

    return {
        "count": len(rows),
        "customers": [row_to_dict(row) for row in rows],
    }


@mcp.tool()
def list_open_tasks(
    customer_id: int | None = None,
    limit: int = 20,
) -> dict[str, Any]:
    """未完了タスクを取得します。"""

    safe_limit = max(1, min(limit, 50))

    with get_connection() as connection:
        if customer_id is None:
            rows = connection.execute(
                """
                SELECT
                    t.id,
                    t.customer_id,
                    c.name AS customer_name,
                    t.title,
                    t.status
                FROM tasks AS t
                JOIN customers AS c
                  ON c.id = t.customer_id
                WHERE t.status = ?
                ORDER BY t.id
                LIMIT ?
                """,
                ("open", safe_limit),
            ).fetchall()
        else:
            rows = connection.execute(
                """
                SELECT
                    t.id,
                    t.customer_id,
                    c.name AS customer_name,
                    t.title,
                    t.status
                FROM tasks AS t
                JOIN customers AS c
                  ON c.id = t.customer_id
                WHERE t.status = ?
                  AND t.customer_id = ?
                ORDER BY t.id
                LIMIT ?
                """,
                ("open", customer_id, safe_limit),
            ).fetchall()

    return {
        "count": len(rows),
        "tasks": [row_to_dict(row) for row in rows],
    }


if __name__ == "__main__":
    mcp.run(transport="stdio")

ここでは、ClaudeからSQLそのものを受け取っていません。

検索条件はSQL文字列へ直接連結せず、プレースホルダーで渡しています。また、検索結果の件数にも上限を設定しています。

20件や50件という数字に特別な意味があるわけではありません。サンプルとして「呼び出し側が無制限に取得できる状態を作らない」ための値です。

6. DB操作ではなく業務操作として公開する

今回のtool名は、

get_customer
search_customers
list_open_tasks

です。

これを、

execute_query
get_table
select_rows

のようなDB寄りの名前にはしていません。

たとえばget_table("customers")を公開すると、customersテーブルの構造そのものが外部インターフェースになりやすくなります。

一方、

SELECT id, name, status
FROM customers

とMCP側で取得列まで決めておけば、あとからテーブルへ別の列が追加されても、自動的にはClaudeへ返りません。

MCPでは「DBの何を触れるか」ではなく、「どの業務操作を許可するか」を先に決めます。

これはセキュリティだけでなく、DBとClaude側を疎結合にする意味でも扱いやすい設計です。

7. Claude Codeへ登録する

Claude CodeからローカルのMCPサーバーへ接続します。

今回はstdioを利用します。

claude mcp add --transport stdio readonly-business-db \
  --scope project \
  -- /absolute/path/to/readonly-business-mcp/.venv/bin/python \
  /absolute/path/to/readonly-business-mcp/server.py

/absolute/path/to/...は自分の環境に置き換えます。

登録後は次のコマンドで確認します。

claude mcp list

接続できたら、たとえば次のように依頼できます。

未完了のタスクを確認してください。
必要な場合だけ業務DBのMCPツールを使ってください。
データに存在しない内容は推測で補わないでください。

ここでも、MCPを接続したからAIの回答が常に正しくなるわけではありません。

DBから取得した情報と、モデルが文章として推論した内容を混同しない設計や運用が必要です。

8. 本番ではDBユーザーも分離する

SQLiteではmode=roを使いました。

実際の業務DBへ接続する場合は、通常の業務システムとMCPでDBユーザーを分ける構成を検討します。

MCP用ユーザーには必要なテーブルやビューへの読み取り権限だけを与えます。

さらに、テーブルを直接参照させず、AI参照用のビューを作る方法もあります。

たとえば顧客テーブルに、

顧客名
ステータス
電話番号
内部メモ
担当者向け情報

があり、Claudeに必要なのが顧客名とステータスだけなら、その2項目だけを公開するビューを作ります。

読み取り専用だから全部読ませてもよい、とは考えません。

社内DBにつなぐ前のチェックリスト

今回のサンプルを実際の業務システムへ広げるなら、私は次のような項目を確認します。

[ ] MCP用DBユーザーを通常アプリと分離した
[ ] MCP用ユーザーに書き込み権限がない
[ ] 必要なテーブル / ビューだけ参照できる
[ ] SELECT * を避けている
[ ] 自由なSQLを受け取るtoolがない
[ ] toolを業務用途ごとに限定している
[ ] 取得件数に上限がある
[ ] SQLへ値を直接文字列連結していない
[ ] AIに不要な列を返していない
[ ] エラーへ秘密情報を含めない
[ ] 必要な認証・認可を検討した
[ ] 必要な監査ログを決めた
[ ] ダミーデータで先に検証した

これはすべてのシステムを網羅するチェックリストではありません。

扱う情報やシステム構成によって必要な対策は変わります。個人情報や機密情報を扱う場合は、組織のルールやセキュリティ要件に合わせた設計が別途必要です。

ハマりどころ

読み取り専用でも、読みすぎることはある

DBへ書き込めないことと、安全に参照できることは同義ではありません。

たとえば一つのテーブルに、通常の業務情報とAIへ渡す必要のない内部情報が混在している場合があります。

そのため、

誰が
どのtoolから
どの列を
何件まで

参照できるのかを考えます。

DBユーザーの権限、参照用ビュー、MCP toolの返却項目は、それぞれ別の層で制限できます。

toolを細かくしすぎない

権限を小さくしようとして、

get_customer_name
get_customer_status
get_customer_id

まで分けると、一つの質問へ答えるために何度もtoolを呼ぶことになります。

反対にexecute_sqlでは広すぎます。

私は「一つの業務上の質問に答えるための自然な単位」を探すのがよいと考えています。

今回のget_customerなら、

{
  "id": 1,
  "name": "サンプル商事",
  "status": "active"
}

程度をまとめて返します。

DBスキーマをそのまま公開しない

自由なSQLをClaudeに生成させる場合、テーブル名やカラム名などを伝える必要が出てきます。

今回の構成では、Claudeが知る必要があるのは、

search_customers(keyword, limit)

というtoolの仕様です。

内部でどのテーブルを使っているかは、通常の利用では意識する必要がありません。

将来DB構造が変わっても、toolの入出力を維持できればClaude側への影響を抑えられます。

MCP toolを、小さなAPIの境界として扱うイメージです。

プロンプトを権限制御の代わりにしない

Claudeへ、

データを変更しないでください。
必要な情報だけ取得してください。

と指示することはできます。

ただ、それだけに依存しません。

人に「この棚だけ見てください」と頼むことと、そもそも別の棚の鍵を渡さないことは違います。

プロンプトは行動方針、権限はシステム側の制約として分けて考えます。

今回なら、プロンプト以前に、

自由なSQL toolを作らない
DBをread-onlyで開く
返す列を限定する
取得件数を制限する

という境界を作っています。

stdioのまま社内共有サービスにはしない

今回の例はローカルで試しやすいようstdioを使っています。

複数ユーザーがネットワーク越しに使うMCPサーバーへ発展させる場合は、認証、認可、通信経路、秘密情報の管理、ログなどを別途設計する必要があります。

transportをHTTP系へ変更するだけで、安全な社内サービスになるわけではありません。

リモート化する場合は、MCP公式仕様のAuthorizationやSecurity Best Practicesも確認して設計します。

まとめ

LLMを社内システムへつなぐとき、最初から「何でもできるAI」を目指すと権限設計が複雑になります。

私はまず読み取り専用にし、さらに読み取り操作自体を限定するところから始めるのが扱いやすいと考えています。

今回作った構成はシンプルです。

Claude
  ↓
MCP Server
  ↓
用途を限定したtool
  ↓
読み取り専用DB

ポイントは、Claudeへ自由なSQL実行権限を渡さなかったことです。

代わりに、

get_customer
search_customers
list_open_tasks

という業務上の操作を公開しました。

この形なら、必要性を確認しながらtoolを一つずつ追加できます。

検索だけで足りるのか。集計が必要なのか。別のデータソースも必要なのか。それを見てから次の権限を考えられます。

私はAIを業務へ入れるとき、「人が決め、AIが下ごしらえする」という役割分担がちょうどいいと考えています。

社内DBとの接続でも、まず必要な範囲だけ読める状態を作る。その境界を確認してから、次に何を任せるかを決める。

MCPはAIへ多くの権限を与えるためだけではなく、AIに何を見せ、何をさせるのかをコードとして明確にする境界にも使えます。

参考文献

10
13
0

Register as a new user and use Qiita more conveniently

  1. You get articles that match your needs
  2. You can efficiently read back useful information
  3. You can use dark theme
What you can do with signing up
10
13

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?