【実装】データベースに接続するMCPサーバー：SQLiteとMySQLで安全なCRUD Toolを作る

1. はじめに：AIにDBを触らせてよいのか

MCP Toolをデータベースへ接続すると、ユーザーは自然言語で次のような操作を依頼できます。

期限が過ぎている未完了タスクを一覧にして
タスク42を完了にして
古いテスト用タスクを削除して

しかし、自然言語からSQLを生成してそのまま実行する設計は危険です。誤った条件のUPDATE、無制限のSELECT、プロンプトインジェクションを起点とする機密データの取得など、SQLインジェクション以外にも多くのリスクがあります。

安全な境界は「AIがSQLを書くこと」ではなく、「AIがサーバー側で許可されたユースケースを選ぶこと」です。上の依頼は、それぞれlist_tasks、update_task、prepare_delete_taskへ割り当てます。SQLとDB認証情報はMCPサーバー内に閉じ込めます。

本記事では、Python 3.12以降、MCP Python SDK v1、SQLite、MySQL Connector/Pythonを使い、安全なCRUD Toolを設計します。2026年7月16日時点でPyPIの最新v1リリースは1.28.1で、v2はプレリリースです。依存関係には意図しないメジャーアップデートを避ける上限を付けます。

dependencies = [
    "mcp[cli]>=1.28,<2",
    "mysql-connector-python",
    "pydantic-settings",
]

2. MCPサーバーからDBへ直接接続する構成

モデルが生成するのはTool名と構造化された引数だけです。MCPサーバーが入力を検証し、固定または許可リストから構築したSQLを実行します。

flowchart LR
    accTitle: MCPデータベース操作の境界
    accDescr: ユーザーの自然言語をAIクライアントがTool呼び出しへ変換し、MCPサーバーが検証済みのSQLだけをSQLiteまたはMySQLへ実行する。認証情報とSQLはサーバー内に保持される。
    U[ユーザー] -->|自然言語| C[AIクライアント]
    C -->|tools/call| M[MCPサーバー]
    M --> V[Schemaと業務検証]
    V --> R[Repository]
    R -->|許可済みSQL| D[(SQLite / MySQL)]
    M -.->|認証情報を渡さない| C

MCPからデータベースまでの安全な処理境界

直接接続したMCPサーバーは、実質的に新しいバックエンドサービスです。認可、監査、エラー処理、接続管理、マイグレーションへの責任を負います。

専用DB、個人用SQLite、読み取り専用分析DB、APIのないPoCには直接接続が適します。一方、既存の業務APIがあるなら原則としてAPIを利用します。MCPだけがDBを直接更新すると、APIが担うバリデーション、ユーザー認可、通知、イベント発行を迂回するためです。

3. 今回作るMCPサーバーの完成形

題材はタスク管理です。tasksに現在の状態を、task_eventsに作成・更新・削除の履歴を保存します。

環境変数DATABASE_BACKEND=sqlite|mysqlでRepositoryを切り替え、ToolのSchemaと名前は変えません。任意SQL Toolは公開せず、更新と監査イベントは同じトランザクションへ入れます。

完成コード：この記事のSQLite・MySQL実装、Docker Compose、テスト一式はGitHubリポジトリで公開しています。以降のコードブロックに表示するファイル名は、このリポジトリ内のパスに対応します。

4. データベース実装をToolから分離する

Tool関数へSQLを直接書くと、入力検証、結果整形、SQL方言、トランザクションが混ざります。間にRepositoryを置きましょう。

from typing import Protocol

class TaskRepository(Protocol):
    def list_tasks(self, query: "ListTasksInput") -> list["Task"]: ...
    def get_task(self, task_id: int) -> "Task | None": ...
    def create_task(self, data: "CreateTaskInput") -> "Task": ...
    def update_task(self, data: "UpdateTaskInput") -> "Task | None": ...
    def delete_task(self, task_id: int, expected_updated_at: str) -> bool: ...

SQLiteTaskRepositoryとMySQLTaskRepositoryがこの契約を実装します。Toolは入力検証と安全な結果への変換に集中でき、SQLのプレースホルダーやlastrowidの差はRepository内へ閉じ込められます。同じ契約テストを両実装へ適用でき、将来ApiTaskRepositoryへ置き換えることも容易です。

5. SQLiteのコネクション管理

小規模なSQLite構成では、Tool呼び出しごとに接続を開閉する方法が明快です。グローバル接続を複数スレッドで共有しません。Pythonのsqlite3はデフォルトで、接続を作成したスレッド以外から利用するとProgrammingErrorを送出します。

from contextlib import contextmanager
import sqlite3

@contextmanager
def sqlite_connection(path: str, *, read_only: bool = False):
    database = f"file:{path}?mode=ro" if read_only else path
    connection = sqlite3.connect(database, uri=read_only)
    connection.row_factory = sqlite3.Row
    try:
        yield connection
        if not read_only:
            connection.commit()
    except Exception:
        if not read_only:
            connection.rollback()
        raise
    finally:
        connection.close()

読み取りToolにはmode=roを使います。このURIで開いた接続に書き込もうとするとSQLite側が拒否するため、コード上の分離だけより強い防御になります。なお、Connectionをコンテキストマネージャーとして使うだけでは接続自体は閉じられないため、明示的なclose()が必要です。

6. MySQLのコネクションプール

MySQLではTool呼び出しごとのTCP接続生成を避け、プールを利用します。読み取り用と更新用でDBユーザーとプールを分けます。

from mysql.connector.pooling import MySQLConnectionPool

reader_pool = MySQLConnectionPool(
    pool_name="mcp_task_reader",
    pool_size=5,
    host=settings.mysql_host,
    user=settings.mysql_reader_user,
    password=settings.mysql_reader_password,
    database=settings.mysql_database,
)

writer_pool = MySQLConnectionPool(
    pool_name="mcp_task_writer",
    pool_size=5,
    host=settings.mysql_host,
    user=settings.mysql_writer_user,
    password=settings.mysql_writer_password,
    database=settings.mysql_database,
)

mcp_readerには必要なテーブルへのSELECTだけを、mcp_writerには必要なSELECT、INSERT、UPDATE、DELETEだけを与えます。管理者権限やDDL権限は不要です。

connection = writer_pool.get_connection()
try:
    # Repository operation
    connection.commit()
except Exception:
    connection.rollback()
    raise
finally:
    connection.close()

プールから取得した接続のclose()は物理切断ではなく、接続をプールへ返す操作です。例外経路でも必ず実行してください。利用可能な接続がなければPoolErrorになるため、プールサイズは同時実行数とDBの接続上限を踏まえて決め、枯渇を監視します。

7. SQLインジェクションを防ぐ

次の文字列連結は使用してはいけません。

sql = f"SELECT * FROM tasks WHERE status = '{status}'"

値は必ずパラメータとしてバインドします。

# SQLite
cursor.execute(
    "SELECT * FROM tasks WHERE status = ?",
    (status,),
)

# MySQL
cursor.execute(
    "SELECT * FROM tasks WHERE status = %s",
    (status,),
)

パラメータ化によりSQLコードとデータが分離されます。OWASPもPrepared Statement／パラメータ化クエリを主要な防御策とし、入力のエスケープだけに依存する方法を強く非推奨としています。

ただし、テーブル名、カラム名、演算子、ASCやDESCは通常プレースホルダーにできません。これらはEnumから定数へマッピングします。

SORT_COLUMNS = {
    TaskSort.CREATED_AT: "created_at",
    TaskSort.DUE_DATE: "due_date",
}
SORT_ORDERS = {
    SortOrder.ASC: "ASC",
    SortOrder.DESC: "DESC",
}

ユーザー入力を「危険な文字がないか」で判定するのではなく、許可された値だけをコード側から選びます。

8. 読み取りToolと更新Toolを分離する

manage_tasks(operation, values)のような汎用Toolは、Schemaが広く、モデルが操作を選びにくく、ログから破壊性も判断しにくくなります。目的別Toolなら入力、権限、監査ラベルを狭くできます。

MCP仕様では、Tool Annotationsはすべてヒントです。信頼できないサーバーからのAnnotationを、クライアントがTool実行判断に利用すべきではないとも明記されています。したがって、destructiveHintは確認UIに役立ちますが、認可やDB権限の代わりにはなりません。

防御は、入力Schema、アプリケーション検証、パラメータ化クエリ、DB権限、トランザクションを重ねて作ります。

9. CRUD Toolを実装する

入力型では、モデルが生成できる値の範囲を狭めます。

from datetime import date
from enum import StrEnum
from typing import Annotated
from pydantic import BaseModel, Field, model_validator

class TaskStatus(StrEnum):
    TODO = "todo"
    IN_PROGRESS = "in_progress"
    DONE = "done"

class ListTasksInput(BaseModel):
    status: TaskStatus | None = None
    limit: Annotated[int, Field(ge=1, le=100)] = 20
    offset: Annotated[int, Field(ge=0)] = 0
    sort_by: TaskSort = TaskSort.CREATED_AT
    sort_order: SortOrder = SortOrder.DESC

class CreateTaskInput(BaseModel):
    title: Annotated[str, Field(min_length=1, max_length=200)]
    description: Annotated[str, Field(max_length=5000)] = ""
    status: TaskStatus = TaskStatus.TODO
    due_date: date | None = None

class UpdateTaskInput(BaseModel):
    task_id: Annotated[int, Field(gt=0)]
    title: Annotated[str, Field(min_length=1, max_length=200)] | None = None
    description: Annotated[str, Field(max_length=5000)] | None = None
    status: TaskStatus | None = None
    due_date: date | None = None

    @model_validator(mode="after")
    def require_change(self):
        if not any(
            value is not None
            for value in (self.title, self.description, self.status, self.due_date)
        ):
            raise ValueError("at least one field must be updated")
        return self

list_tasksは最大100件に制限し、ソート列を許可リストから選びます。get_taskの不存在はDB障害ではなく、{"found": false}のような正常な結果として返します。内部例外はログへ残しても、SQL文、ファイルパス、ホスト名、認証情報をTool結果へ含めません。

create_taskではINSERT後のlastrowidを使い、同じ接続上で作成済みレコードを取得します。update_taskは自由なdictを受け取らず、更新可能な列だけを型として公開します。

@mcp.tool(annotations={"readOnlyHint": False, "destructiveHint": True})
def update_task(input: UpdateTaskInput) -> dict:
    task = repository.update_task(input)
    if task is None:
        return {
            "updated": False,
            "reason": "not_found_or_conflict",
            "task_id": input.task_id,
        }
    return {"updated": True, "task": task.model_dump(mode="json")}

競合を正確に区別したい場合は、入力にexpected_updated_atまたはバージョン番号を加え、WHERE id = ? AND updated_at = ?で楽観ロックを行います。rowcount == 0なら再取得し、不存在と更新競合を分類できます。

10. トランザクションとロールバック

update_taskでは、タスク本体の更新と監査イベントの追加を同じトランザクションに含めます。

try:
    updated = update_task_row(connection, input)
    if not updated:
        raise TaskConflictError(input.task_id)

    insert_task_event(
        connection,
        task_id=input.task_id,
        event_type="updated",
        event_data=changes_json,
    )
    connection.commit()
except Exception:
    connection.rollback()
    raise

履歴INSERTに失敗した場合、タスク更新もロールバックされます。途中でcommit()してはいけません。MySQL Connector/Pythonはデフォルトでautocommitが無効なので、成功時の明示的なcommit()と失敗時のrollback()が必要です。SQLiteでもRepositoryの接続コンテキストに同じ境界を持たせます。

トランザクション管理はToolではなくRepository側に置くと、監査イベントを書き忘れにくくなります。また、メール送信や外部API呼び出しをDBトランザクション内へ入れるとロック時間が延び、外部処理だけ成功する問題も生じます。必要ならoutboxパターンでトランザクション後に処理します。

11. 削除操作に確認を入れる

confirm=Trueだけでは確認になりません。モデル自身がBooleanを生成できるからです。直前に確認した対象、レコードの版、期限、操作を署名付きトークンへ結び付けます。

sequenceDiagram
    accTitle: 署名付きトークンによる削除確認
    accDescr: AIクライアントが削除準備Toolを呼び、サーバーが対象情報と短時間有効な署名済みトークンを返す。ユーザー承認後、削除Toolが署名、期限、対象、更新時刻、未使用状態を検証し、削除と監査記録を同じトランザクションで実行する。
    actor U as ユーザー
    participant C as AIクライアント
    participant M as MCPサーバー
    participant D as データベース
    U->>C: タスク42を削除して
    C->>M: prepare_delete_task(42)
    M->>D: タスクとupdated_atを取得
    D-->>M: 対象レコード
    M-->>C: 対象情報と署名済みトークン
    C-->>U: このタスクを削除しますか？
    U->>C: 削除して
    C->>M: confirm_delete_task(42, token)
    M->>M: 署名・期限・ID・版・nonceを検証
    M->>D: DELETEと監査INSERT
    D-->>M: COMMIT
    M-->>C: 削除完了

削除を確認する2段階シーケンス

トークンのペイロードにはoperation=delete_task、task_id、updated_at、有効期限exp、ランダムなnonceを含め、サーバー秘密鍵でHMAC署名します。confirm_delete_taskは次を検証します。

• 署名が正しく、有効期限内である
• 引数とトークンのタスクIDが一致する
• 現在のupdated_atが発行時と一致する
• 操作種別がdelete_taskである
• nonceが未使用である

改変、期限切れ、別タスクへの流用、確認後に変更されたレコード、再実行を拒否します。nonceの消費、削除、task_eventsへのdeletedイベント追加は同じトランザクションで行います。トークンをログへ平文で残さず、秘密鍵は環境変数またはシークレット管理基盤から読み込みます。

12. 任意SQL実行Toolを公開してはいけない理由

次のToolは、DBドライバーをMCPへ露出しているだけです。

@mcp.tool()
def execute_sql(sql: str) -> list[dict]:
    ...

このSchemaでは操作範囲を制限できません。DROPや無条件DELETEだけでなく、大量取得、機密列やシステムテーブルの参照、重いJOIN、ロックを伴うクエリも許してしまいます。Tool名からリスクを判定できず、監査ログも「SQL実行」としか分類できません。

「SELECTだけ」も十分ではありません。SQLの構文解析で完全に読み取り専用を保証するのは容易でなく、読み取りであっても情報漏えいと可用性低下を起こせます。代わりにget_overdue_tasks、count_tasks_by_status、get_task_activityのような目的別Toolを作り、列、件数、条件、実行時間をサーバー側で制御します。

13. 安全性をテストする

Repository契約テストをSQLiteとMySQLの両方へ適用します。最低限、次を確認します。

• 存在する／存在しないタスクを正しく扱う
• 不正なステータスとlimit=1000をSchemaで拒否する
• 作成・更新・削除イベントが保存される
• DB例外に内部情報を含めずToolエラーへ変換する
• 読み取り用接続で書き込みが失敗する

SQLインジェクション試験では、タイトルや検索値へ' OR 1=1 --を渡します。文字列がデータとして保存・比較され、SQL構造を変えないことを確認します。危険文字を拒否する試験ではなく、正当な文字列として安全に扱えることの試験です。

トランザクション試験では、タスク更新後の履歴INSERTを意図的に失敗させ、タスク本体も元の状態へ戻ることを確認します。削除では、改変・期限切れ・別タスク用・発行後に更新されたトークンと、同一トークンの再利用を試します。

MySQL統合テストはDocker Composeで実DBを起動し、SQLiteと同じ契約テストを実行します。プレースホルダー、日時型、rowcount、トランザクションの違いはモックだけでは見落としやすいためです。

14. AIクライアントから実行する

MCP Inspectorまたは対応クライアントで、自然言語が期待したTool呼び出しへ変換されることを確認します。

ユーザー: 未完了で期限が近いタスクを5件表示して
AI: list_tasks(status="todo", limit=5, sort_by="due_date", sort_order="asc")

ユーザー: タスク12を完了にして
AI: update_task(task_id=12, status="done")

ユーザー: タスク42を削除して
AI: prepare_delete_task(task_id=42)
AI: 「古いテストタスク」を削除しますか？
ユーザー: 削除して
AI: confirm_delete_task(task_id=42, confirmation_token="...")

Tool選択が正しくても、それだけで認可済みとは限りません。リモートMCPサーバーでは、認証された主体がそのタスクを操作できるかをサーバー側で確認します。Tool入力のuser_idを信用せず、認証コンテキストから主体を取得してください。

15. 本番利用に向けたチェックリストとまとめ

• 任意SQL Toolを公開していない
• SQLへ入力値を文字列連結していない
• 型、文字数、件数、ソート項目に制限がある
• 認証された主体に対する認可をTool内で検証している
• 読み取り用と更新用の接続・DB権限を分けている
• DB管理者アカウントやDDL権限を使用していない
• 接続とカーソルを例外時にも閉じている
• 複数更新を1トランザクションとして扱っている
• エラー結果へSQLや認証情報を含めていない
• 破壊的操作を対象・版・期限と結び付けて確認している
• 監査イベントとTool呼び出しを追跡できる
• レート制限、タイムアウト、取得件数上限がある
• 既存の業務APIがある場合はAPI経由を優先している

MCP Toolはデータベース操作の薄いラッパーではなく、権限とユースケースの境界です。AIへSQLを渡すのではなく、get_taskやupdate_taskという限定された能力を渡します。そのうえでSchema、業務検証、パラメータ化、最小権限、トランザクション、監査、ユーザー確認を重ねます。

SQLiteからMySQLへ移行しても、この境界は変わりません。変えるべきなのは接続管理とSQL方言であり、AIへ許可する操作の範囲ではありません。