【次世代AI開発：第2回】Claude・Cursorと自社DBをつなぐ実践MCPサーバー構築

【新連載：MCP・高度RAG・自律エージェントの現場設計論】

前回（第1回）：【次世代AI開発：第1回】MCP（Model Context Protocol）の内部構造解剖と自作サーバー実装

※本連載は、MCP（Model Context Protocol）、Context Engineering、GraphRAG、Self-Healing Agent、ローカルSLMなどの最先端技術スタックを用い、実務プロダクションで真に耐えうる次世代AIシステムを構築する重厚なハンズオン連載です。

1. イントロダクション（本記事のねらいと到達目標）

前回は、MCP (Model Context Protocol) の内部仕様である JSON-RPC 2.0 通信メカニズムと、FastMCP フレームワークを使った基本的な Tools / Resources / Prompts の実装方法を解剖しました。

しかし、実際のシステム開発や日常の開発業務において真に価値を発揮するのは、「普段使っているエージェント（Cursor, Windsurf, Claude Desktop）から、社内データベースや固有のREST APIへ安全かつ型安全にアクセスさせる」実戦的な組み込みです。

例えば、AIエージェントに「直近の売上低下の原因となっている顧客IDと注文ログを調査して」と指示した際、エージェントが自律的に社内DBを安全に検索し、結果をまとめてコード補完やレポート作成を行ってくれれば、開発生産性は飛躍的に向上します。

第2回となる今回は、LLMに直接的な危険なSQLを実行させず、パラメータ化された型安全な関数（Tools）を介して社内データベース（SQLite / PostgreSQL想定）とセキュアにやり取りする実用的なMCPサーバーを構築します。さらに、Cursor（.cursor/mcp.json）および Claude Desktop への実際の組み込み・運用設定手順までを完全解説することを目標とします。

前提条件と動作検証環境（Prerequisites）

Pythonバージョン: Python 3.10 以上
動作確認済み主要ライブラリ:
- mcp>=1.2.0（Anthropic公式 MCP SDK）
- sqlite3（Python標準ライブラリ）
対象クライアント環境:
- Cursor Editor（v0.45以上、MCPサポート版）
- Claude Desktop（macOS / Windows）

2. 【理論・背景】エージェント×データベース接続におけるセキュリティ課題

LLMをデータベース（DB）に接続させるアプローチには、大きく分けて2つの設計パターンが存在します。

1. 危険なアンチパターン：直接的なText-to-SQL実行

LLMにDBスキーマを渡し、動的に生成された生のSQL文字列（SELECT * FROM ..., DELETE FROM ...）をそのままDBに対して発行させる方式です。

【危険なアプローチ】
[ユーザー指示] ---> [LLM] ---> 生のSQL文 `DROP TABLE users;` ---> [本番DB (破壊)]

リスク1: SQLインジェクションと破壊的操作: プロンプトインジェクション等により、データ消去や構造破壊コマンドが実行される危険性があります。
リスク2: 情報漏洩: 権限外のテーブル（パスワードハッシュ、個人情報など）を自由検索される恐れがあります。

2. 安全なMCPアプローチ：型安全な定型Toolへのカプセル化

MCPを介した設計では、生のSQL実行を一切許可せず、パラメータ化された定型クエリ関数（Tools） のみをMCPサーバー経由で露出させます。

【セキュアなMCPアプローチ】
[ユーザー指示] ---> [LLM] ---> JSONパラメータ `{"user_id": "usr_101"}` 
                                   ↓ (JSON-RPC)
                              [MCP Server] ---> パラメータ化クエリ ---> [DB (安全)]

3. DB接続における2つのアプローチの比較

比較項目	生のText-to-SQL直接実行	MCPによるツールカプセル化 (本手法)
セキュリティ	極めて危険（SQLインジェクション、データ消去リスク）	極めて安全（パラメータ化クエリでSQL固定化）
型安全性	なし（LLMの出力文字列に依存）	あり（Pydantic/Python型ヒントで完全バリデーション）
レスポンス制御	DB全体のスキーマ理解が必要でトークン浪費	必要なデータのみを最小限のJSON/Textで返却可能
アクセス権限	DB接続アカウントの権限に依存	露出するTool関数単位で読み書き権限を厳密制御可能

3. 【アーキテクチャ解剖】Cursor / Claude と自社DB MCPサーバーのデータフロー

Cursor や Claude Desktop がローカルの Python MCP サーバーを介してデータベースと対話する内部通信フローは以下の通りです。

4. 【完全実装】自社DBと連携する型安全な実用MCPサーバー

それでは、実際に顧客情報テーブルと注文履歴テーブルを持つ SQLite データベースを自動構築し、Cursor や Claude から安全に呼び出せるMCPサーバー mcp_db_server.py を作成しましょう。

新規ファイル mcp_db_server.py を作成し、以下の完全なコードを記述します。

# 動作確認済みライブラリバージョン: mcp>=1.2.0
import os
import sys
import sqlite3
from typing import List, Dict, Any, Optional
from mcp.server.fastmcp import FastMCP

# 1. DBファイルとMCPサーバーの初期化
DB_FILE = "company_business.db"
mcp = FastMCP("CompanyDatabaseServer")


# =====================================================================
# データベースの初期化（サンプルテーブルとデモデータの自動生成）
# =====================================================================
def init_database():
    """DBが存在しない場合、テーブルとデモデータを自動生成します。"""
    conn = sqlite3.connect(DB_FILE)
    cursor = conn.cursor()
    
    # 顧客テーブル
    cursor.execute("""
    CREATE TABLE IF NOT EXISTS customers (
        customer_id TEXT PRIMARY KEY,
        name TEXT NOT NULL,
        email TEXT NOT NULL,
        plan TEXT NOT NULL
    );
    """)
    
    # 注文履歴テーブル
    cursor.execute("""
    CREATE TABLE IF NOT EXISTS orders (
        order_id TEXT PRIMARY KEY,
        customer_id TEXT NOT NULL,
        product_name TEXT NOT NULL,
        amount INTEGER NOT NULL,
        status TEXT NOT NULL,
        created_at TEXT NOT NULL,
        FOREIGN KEY (customer_id) REFERENCES customers (customer_id)
    );
    """)
    
    # サンプルデータの挿入（テーブルが空の場合のみ）
    cursor.execute("SELECT COUNT(*) FROM customers;")
    if cursor.fetchone()[0] == 0:
        cursor.executemany(
            "INSERT INTO customers VALUES (?, ?, ?, ?);",
            [
                ("usr_101", "山田 太郎", "yamada@example.com", "Enterprise"),
                ("usr_102", "佐藤 花子", "sato@example.com", "Pro"),
                ("usr_103", "鈴木 一郎", "suzuki@example.com", "Free"),
            ]
        )
        cursor.executemany(
            "INSERT INTO orders VALUES (?, ?, ?, ?, ?, ?);",
            [
                ("ord_9001", "usr_101", "クラウドサーバー 月額", 50000, "COMPLETED", "2026-07-01 10:00:00"),
                ("ord_9002", "usr_101", "データベースバックアップ オプション", 12000, "COMPLETED", "2026-07-05 14:30:00"),
                ("ord_9003", "usr_102", "Proプラン 年間ライセンス", 36000, "COMPLETED", "2026-07-10 09:15:00"),
                ("ord_9004", "usr_101", "AI分析コンサルティング", 150000, "PENDING", "2026-07-20 16:00:00"),
            ]
        )
        conn.commit()
    conn.close()

# 起動時にDB初期化を実行
init_database()


# =====================================================================
# Tools の定義（パラメータ化クエリによる安全な関数）
# =====================================================================

@mcp.tool()
def search_customer_by_id(customer_id: str) -> str:
    """指定された顧客ID（例: usr_101）から顧客の基本情報を取得します。

    Args:
        customer_id (str): 顧客ID（例: 'usr_101'）
    """
    conn = sqlite3.connect(DB_FILE)
    conn.row_factory = sqlite3.Row
    cursor = conn.cursor()
    
    # 安全なパラメータ化クエリ（プレースホルダー ? を使用）
    cursor.execute("SELECT * FROM customers WHERE customer_id = ?;", (customer_id,))
    row = cursor.fetchone()
    conn.close()
    
    if not row:
        return f"検索結果: 顧客ID '{customer_id}' は存在しません。"
        
    return f"【顧客情報】\n- ID: {row['customer_id']}\n- 氏名: {row['name']}\n- Email: {row['email']}\n- 契約プラン: {row['plan']}"


@mcp.tool()
def list_orders_for_customer(customer_id: str, limit: int = 5) -> str:
    """指定された顧客IDの注文履歴一覧を取得します（最新順）。

    Args:
        customer_id (str): 顧客ID（例: 'usr_101'）
        limit (int): 取得する最大件数（デフォルト: 5）
    """
    conn = sqlite3.connect(DB_FILE)
    conn.row_factory = sqlite3.Row
    cursor = conn.cursor()
    
    cursor.execute(
        "SELECT * FROM orders WHERE customer_id = ? ORDER BY created_at DESC LIMIT ?;",
        (customer_id, limit)
    )
    rows = cursor.fetchall()
    conn.close()
    
    if not rows:
        return f"検索結果: 顧客ID '{customer_id}' の注文履歴は見つかりませんでした。"
        
    result_text = f"【顧客 {customer_id} の注文履歴 (全{len(rows)}件)】\n"
    total_amount = 0
    for r in rows:
        result_text += f"- 注文ID: {r['order_id']} | 商品: {r['product_name']} | 金額: ¥{r['amount']:,} | ステータス: {r['status']} | 日時: {r['created_at']}\n"
        total_amount += r['amount']
        
    result_text += f"└ 合計金額: ¥{total_amount:,}"
    return result_text


@mcp.tool()
def create_new_order(customer_id: str, product_name: str, amount: int) -> str:
    """新規の注文レコードを安全に追加作成します。

    Args:
        customer_id (str): 注文を行う顧客ID
        product_name (str): 購入する商品・サービス名
        amount (int): 金額（日本円）
    """
    if amount <= 0:
        return "エラー: 注文金額は1円以上である必要があります。"
        
    conn = sqlite3.connect(DB_FILE)
    cursor = conn.cursor()
    
    # 顧客の存在確認
    cursor.execute("SELECT customer_id FROM customers WHERE customer_id = ?;", (customer_id,))
    if not cursor.fetchone():
        conn.close()
        return f"エラー: 指定された顧客ID '{customer_id}' が存在しないため注文を作成できません。"
        
    # 新規注文IDの生成
    import random
    new_order_id = f"ord_{random.randint(9005, 9999)}"
    now_str = datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S")
    
    cursor.execute(
        "INSERT INTO orders VALUES (?, ?, ?, ?, ?, ?);",
        (new_order_id, customer_id, product_name, amount, "PENDING", now_str)
    )
    conn.commit()
    conn.close()
    
    return f"✅ 新規注文を作成しました (注文ID: {new_order_id}, 顧客: {customer_id}, 商品: {product_name}, 金額: ¥{amount:,})"


# =====================================================================
# Resources の定義（データベース統計メタデータの露出）
# =====================================================================

@mcp.resource("db://schema/summary")
def get_db_schema_summary() -> str:
    """データベースのテーブル構造と統計サマリーを返す静的リソース"""
    conn = sqlite3.connect(DB_FILE)
    cursor = conn.cursor()
    
    cursor.execute("SELECT COUNT(*) FROM customers;")
    customer_count = cursor.fetchone()[0]
    
    cursor.execute("SELECT COUNT(*) FROM orders;")
    order_count = cursor.fetchone()[0]
    conn.close()
    
    return f"""
[社内データベース構造定義]
- 顧客テーブル (customers): customer_id, name, email, plan
- 注文テーブル (orders): order_id, customer_id, product_name, amount, status, created_at
- 登録顧客数: {customer_count} 名
- 総注文数: {order_count} 件
"""


if __name__ == "__main__":
    # Stdioトランスポートで起動
    print("🚀 DB MCP サーバーを起動中...", file=sys.stderr)
    mcp.run()

5. コードの行別・ロジック詳細解説

実装した mcp_db_server.py の技術的な工夫点とポイントを解説します。

1. `sqlite3.Row` による辞書風アクセスとパラメータ化クエリ

conn.row_factory = sqlite3.Row を指定することで、クエリ結果を r['customer_id'] のようにカラム名で安全に参照可能にしています。
全すべてのSQL実行（cursor.execute）において、プレースホルダー ? を使用した パラメータ化クエリ を徹底しています。これにより、LLMが万が一悪意ある文字列を生成した場合でも、単なる値データとしてエスケープされ、SQLインジェクションが物理的に不可能です。

2. トランザクション処理とクリーンアップ

書き込み操作（create_new_order）では、必ず conn.commit() を呼び出して変更を保存し、各関数内で確実に conn.close() を呼び出してデータベースのロックを解除しています。

3. 明確なバリデーションエラーの返却

入力値（amount <= 0 や顧客の未存在）をPythonロジック側で検証し、明確なエラーテキストを返却することで、LLMが自律的に理由を理解して修正発言を行えるように設計しています。

6. プロダクション導入・クライアント設定・セキュリティ最適化

作成した mcp_db_server.py を、Cursor および Claude Desktop に登録して実際にエージェントから呼び出す設定手順を解説します。

1. Cursor エディタへの設定手順（プロジェクト別 / グローバル）

Cursor では、設定画面のUIから登録するか、プロジェクト直下の .cursor/mcp.json ファイルを作成して設定します。

プロジェクトローカル設定（`.cursor/mcp.json`）

プロジェクトのルートディレクトリに .cursor フォルダを作成し、以下の mcp.json を配置します。

{
  "mcpServers": {
    "company_db": {
      "command": "python",
      "args": [
        "N:/antigravity/qiita/mcp_db_server.py"
      ]
    }
  }
}

設定の適用確認

Cursorの 「Cursor Settings」 -> 「MCP」 を開きます。
company_db が表示され、緑色のランプ（Active）が点灯していることを確認します。
Cursor Chat（Ctrl + L や Cmd + L）を開き、「顧客 usr_101 の注文履歴を調べて」と指示すると、Cursorが自動で list_orders_for_customer ツールを呼び出して結果を返してくれます。

【Cursor Chatでの対話例】
ユーザー: 顧客 usr_101 の直近の注文内容と合計金額を教えて。
Cursor: ツール `company_db:list_orders_for_customer` を呼び出します...
        (実行成功)
        usr_101（山田 太郎様）の直近の注文履歴は以下の3件で、合計金額は212,000円です。
        - クラウドサーバー 月額 (¥50,000)
        - データベースバックアップ (¥12,000)
        - AI分析コンサルティング (¥150,000)

2. Claude Desktop への設定手順

claude_desktop_config.json（Windows: %APPDATA%\Claude\claude_desktop_config.json / Mac: ~/Library/Application Support/Claude/claude_desktop_config.json）を開き、以下を追加します。

{
  "mcpServers": {
    "company_db": {
      "command": "python",
      "args": [
        "C:/path/to/mcp_db_server.py"
      ]
    }
  }
}

3. セキュリティ最適化のベストプラクティス

読み取り専用（Read-only）モードの分離: 本番運用においては、参照用ツール（search 等）のみを搭載したMCPサーバーと、書き込み用ツール（create 等）を持つMCPサーバーを分け、環境変数 READ_ONLY=true で書き込み機能をスイッチ化する設計が推奨されます。
絶対パスの利用: mcp.json 内の command や args に記述するPythonパスおよびスクリプトパスは、相対パスによる誤動作を防ぐため、必ず 絶対パス で指定してください。

7. まとめと次回の展望

今回は、エージェント（Cursor / Claude Desktop）から社内データベースに型安全かつセキュアにアクセスするための実用的なMCPサーバーを構築し、実際の開発環境へ組み込んで動作させる手順を解説しました。

生のSQLを渡さず、パラメータ化された定型ツールとしてカプセル化することで、セキュリティリスクを皆無にしながらエージェントの作業能力を何倍にも引き出せることを体感できたかと思います。

しかし、エージェントが複雑なタスクをこなすようになると、今度は「LLMに渡すプロンプトや文脈（コンテキスト）の肥大化」という次の壁にぶつかります。トークンコストが爆発し、長文に埋もれて精度が低下する「Lost in the Middle」現象です。

次回、第3回。
プロンプトエンジニアリングの枠を超え、長文LLM時代におけるコンテキスト管理の決定版、**『PromptからContextへ：長文LLM時代の Context Engineering の極意』**に進みます。

トークンコストを最小化しつつ、エージェントの記憶と文脈を最適制御する最先端のコンテキストエンジニアリングへ進みましょう。