はじめに

この記事は、QiitaのModel Context Protocol（以下、MCP）解説シリーズの第15回です。

今回は、MCPをより実用的に活用するための応用編として、データベース連携に挑戦します。特に、手軽で軽量なSQLiteデータベースを読み取り専用で安全に公開する方法を解説します。

💡 なぜデータベース連携が重要なのか？

MCPの強みは、LLMが外部データにアクセスできることです。これまではテキストファイルや静的なデータを使ってきましたが、実際のアプリケーションでは、頻繁に更新される情報や、複雑な構造を持つデータを扱う必要があります。

データベース連携のメリット

リアルタイム性の確保: 最新の在庫情報、顧客データ、売上データなど、常に変動する情報をLLMに提供できます
効率的なデータアクセス: 必要な情報だけをクエリで取得できるため、メモリや帯域幅を節約し、LLMの応答速度を向上させます
データの構造化: データベースは情報を構造化して格納するため、LLMがデータを正確に理解し、利用しやすくなります
スケーラビリティ: 大量のデータを効率的に処理できます

🛠️ ステップ1：プロジェクトのセットアップとデータベースの準備

今回はTypeScriptを使って実装します。一貫性を保つため、これまでのシリーズと同じ言語を使用します。

プロジェクトの初期化

mkdir mcp-sqlite-example
cd mcp-sqlite-example
npm init -y
npm install @modelcontextprotocol/sdk sqlite3 zod
npm install -D typescript ts-node @types/node @types/sqlite3
npx tsc --init

サンプルデータベースの作成

create-database.jsファイルを作成してサンプルデータを準備します：

const sqlite3 = require('sqlite3').verbose();

// データベースファイルを作成
const db = new sqlite3.Database('products.db');

db.serialize(() => {
  // テーブルの作成
  db.run(`CREATE TABLE IF NOT EXISTS products (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    name TEXT NOT NULL,
    category TEXT NOT NULL,
    price REAL NOT NULL,
    description TEXT,
    stock_quantity INTEGER DEFAULT 0,
    created_at DATETIME DEFAULT CURRENT_TIMESTAMP
  )`);

  db.run(`CREATE TABLE IF NOT EXISTS categories (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    name TEXT UNIQUE NOT NULL,
    description TEXT
  )`);

  db.run(`CREATE TABLE IF NOT EXISTS orders (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    customer_name TEXT NOT NULL,
    product_id INTEGER,
    quantity INTEGER NOT NULL,
    order_date DATETIME DEFAULT CURRENT_TIMESTAMP,
    FOREIGN KEY (product_id) REFERENCES products (id)
  )`);

  // サンプルデータの挿入
  const categories = [
    ['Electronics', '電子機器・デバイス'],
    ['Accessories', 'アクセサリー・周辺機器'],
    ['Software', 'ソフトウェア・アプリケーション']
  ];

  categories.forEach(([name, description]) => {
    db.run('INSERT OR IGNORE INTO categories (name, description) VALUES (?, ?)', [name, description]);
  });

  const products = [
    ['Laptop Pro 15"', 'Electronics', 1299.99, 'High-performance laptop with 16GB RAM', 25],
    ['Wireless Keyboard', 'Accessories', 79.99, 'Ergonomic wireless keyboard with backlight', 50],
    ['Optical Mouse', 'Accessories', 29.99, 'Precision optical mouse with programmable buttons', 75],
    ['Code Editor Pro', 'Software', 99.99, 'Professional code editor with advanced features', 100],
    ['Gaming Headset', 'Accessories', 149.99, 'High-quality gaming headset with surround sound', 30],
    ['Tablet 10"', 'Electronics', 449.99, 'Lightweight tablet with high-resolution display', 40]
  ];

  products.forEach(([name, category, price, description, stock]) => {
    db.run('INSERT INTO products (name, category, price, description, stock_quantity) VALUES (?, ?, ?, ?, ?)', 
           [name, category, price, description, stock]);
  });

  const orders = [
    ['Alice Johnson', 1, 1],
    ['Bob Smith', 2, 2],
    ['Charlie Brown', 3, 1],
    ['Diana Wilson', 1, 1],
    ['Eve Davis', 4, 3]
  ];

  orders.forEach(([customer, productId, quantity]) => {
    db.run('INSERT INTO orders (customer_name, product_id, quantity) VALUES (?, ?, ?)', 
           [customer, productId, quantity]);
  });
});

db.close((err) => {
  if (err) {
    console.error('データベース作成エラー:', err);
  } else {
    console.log('データベース products.db が正常に作成されました！');
  }
});

データベースを作成：

node create-database.js

📝 ステップ2：MCPサーバーの実装

server.tsファイルを作成し、以下のコードを実装します：

#!/usr/bin/env node
import { Server } from "@modelcontextprotocol/sdk/server/index.js";
import { StdioServerTransport } from "@modelcontextprotocol/sdk/server/stdio.js";
import {
  CallToolRequestSchema,
  ListResourcesRequestSchema,
  ListToolsRequestSchema,
  ReadResourceRequestSchema,
  TextContent,
} from "@modelcontextprotocol/sdk/types.js";
import { z } from "zod";
import sqlite3 from "sqlite3";
import { promisify } from "util";

// SQLインジェクション防止のための安全なクエリチェック
class QueryValidator {
  private static readonly ALLOWED_KEYWORDS = [
    'SELECT', 'FROM', 'WHERE', 'JOIN', 'INNER', 'LEFT', 'RIGHT', 'OUTER',
    'ON', 'AND', 'OR', 'GROUP', 'BY', 'HAVING', 'ORDER', 'LIMIT', 'OFFSET',
    'AS', 'DISTINCT', 'COUNT', 'SUM', 'AVG', 'MIN', 'MAX', 'CASE', 'WHEN',
    'THEN', 'ELSE', 'END', 'IN', 'LIKE', 'BETWEEN', 'IS', 'NULL', 'NOT'
  ];

  private static readonly FORBIDDEN_KEYWORDS = [
    'INSERT', 'UPDATE', 'DELETE', 'DROP', 'CREATE', 'ALTER', 'TRUNCATE',
    'REPLACE', 'MERGE', 'GRANT', 'REVOKE', 'COMMIT', 'ROLLBACK', 'PRAGMA'
  ];

  static validateReadOnlyQuery(query: string): void {
    const upperQuery = query.trim().toUpperCase();
    
    // SELECT文で始まることを確認
    if (!upperQuery.startsWith('SELECT')) {
      throw new Error('読み取り専用クエリのみ許可されています。SELECT文を使用してください。');
    }

    // 禁止されたキーワードがないかチェック
    for (const forbidden of this.FORBIDDEN_KEYWORDS) {
      if (upperQuery.includes(forbidden)) {
        throw new Error(`禁止されたSQL操作が含まれています: ${forbidden}`);
      }
    }

    // セミコロンで複数のクエリを実行することを防ぐ
    const statements = query.split(';').filter(s => s.trim());
    if (statements.length > 1) {
      throw new Error('複数のSQL文の実行は許可されていません。');
    }
  }
}

// データベース接続の管理
class DatabaseManager {
  private db: sqlite3.Database;

  constructor(dbPath: string) {
    this.db = new sqlite3.Database(dbPath, sqlite3.OPEN_READONLY, (err) => {
      if (err) {
        console.error('データベース接続エラー:', err);
        throw err;
      }
      console.error('SQLiteデータベースに正常に接続しました');
    });
  }

  async query(sql: string): Promise<any[]> {
    return new Promise((resolve, reject) => {
      this.db.all(sql, (err, rows) => {
        if (err) {
          reject(err);
        } else {
          resolve(rows);
        }
      });
    });
  }

  async getTableSchema(tableName: string): Promise<any[]> {
    return new Promise((resolve, reject) => {
      this.db.all(`PRAGMA table_info(${tableName})`, (err, rows) => {
        if (err) {
          reject(err);
        } else {
          resolve(rows);
        }
      });
    });
  }

  async getTableNames(): Promise<string[]> {
    return new Promise((resolve, reject) => {
      this.db.all(
        "SELECT name FROM sqlite_master WHERE type='table' ORDER BY name",
        (err, rows: any[]) => {
          if (err) {
            reject(err);
          } else {
            resolve(rows.map(row => row.name));
          }
        }
      );
    });
  }

  close(): void {
    this.db.close();
  }
}

// 入力スキーマの定義
const QueryDatabaseSchema = z.object({
  query: z.string().min(1).describe("実行するSQL SELECTクエリ")
});

const GetTableSchemaSchema = z.object({
  tableName: z.string().min(1).describe("スキーマを取得するテーブル名")
});

class SQLiteMCPServer {
  private server: Server;
  private dbManager: DatabaseManager;

  constructor(dbPath: string) {
    this.dbManager = new DatabaseManager(dbPath);
    
    this.server = new Server(
      {
        name: "sqlite-mcp-server",
        version: "1.0.0",
      },
      {
        capabilities: {
          tools: {},
          resources: {},
        },
      }
    );

    this.setupHandlers();
  }

  private setupHandlers() {
    // ツール一覧の取得
    this.server.setRequestHandler(ListToolsRequestSchema, async () => ({
      tools: [
        {
          name: "query_database",
          description: "SQLiteデータベースに対してSELECTクエリを実行し、結果を取得する",
          inputSchema: {
            type: "object",
            properties: {
              query: {
                type: "string",
                description: "実行するSQL SELECTクエリ（読み取り専用）"
              }
            },
            required: ["query"]
          }
        },
        {
          name: "get_table_schema",
          description: "指定されたテーブルのスキーマ情報を取得する",
          inputSchema: {
            type: "object",
            properties: {
              tableName: {
                type: "string",
                description: "スキーマを取得するテーブル名"
              }
            },
            required: ["tableName"]
          }
        },
        {
          name: "list_tables",
          description: "データベース内の全テーブル名を取得する",
          inputSchema: {
            type: "object",
            properties: {},
            required: []
          }
        }
      ]
    }));

    // ツール実行の処理
    this.server.setRequestHandler(CallToolRequestSchema, async (request) => {
      const { name, arguments: args } = request.params;

      try {
        switch (name) {
          case "query_database":
            return await this.handleQueryDatabase(args);
          case "get_table_schema":
            return await this.handleGetTableSchema(args);
          case "list_tables":
            return await this.handleListTables();
          default:
            throw new Error(`未知のツール: ${name}`);
        }
      } catch (error) {
        const errorMessage = error instanceof Error ? error.message : "不明なエラー";
        return {
          content: [
            {
              type: "text",
              text: `❌ エラー: ${errorMessage}`
            } as TextContent
          ],
          isError: true
        };
      }
    });

    // リソース一覧の取得
    this.server.setRequestHandler(ListResourcesRequestSchema, async () => {
      try {
        const tables = await this.dbManager.getTableNames();
        return {
          resources: tables.map(tableName => ({
            uri: `sqlite:///${tableName}`,
            name: `Table: ${tableName}`,
            description: `SQLiteテーブル ${tableName} のスキーマ情報`,
            mimeType: "application/json"
          }))
        };
      } catch (error) {
        console.error('リソース一覧取得エラー:', error);
        return { resources: [] };
      }
    });

    // リソース読み取りの処理
    this.server.setRequestHandler(ReadResourceRequestSchema, async (request) => {
      const { uri } = request.params;
      
      try {
        if (!uri.startsWith('sqlite:///')) {
          throw new Error('サポートされていないURIスキーム');
        }

        const tableName = uri.replace('sqlite:///', '');
        const schema = await this.dbManager.getTableSchema(tableName);
        
        return {
          contents: [
            {
              uri,
              mimeType: "application/json",
              text: JSON.stringify({
                tableName,
                schema: schema,
                description: `テーブル ${tableName} のスキーマ情報`
              }, null, 2)
            }
          ]
        };
      } catch (error) {
        throw new Error(`リソース読み取りエラー: ${error instanceof Error ? error.message : '不明なエラー'}`);
      }
    });
  }

  private async handleQueryDatabase(args: any) {
    const parsed = QueryDatabaseSchema.parse(args);
    
    try {
      // クエリの安全性をチェック
      QueryValidator.validateReadOnlyQuery(parsed.query);
      
      // クエリの実行
      const results = await this.dbManager.query(parsed.query);
      
      const resultText = results.length === 0 
        ? "クエリの実行は成功しましたが、結果は0件でした。"
        : `クエリの実行結果（${results.length}件）:\n\n${JSON.stringify(results, null, 2)}`;

      return {
        content: [
          {
            type: "text",
            text: resultText
          } as TextContent
        ]
      };
    } catch (error) {
      const errorMessage = error instanceof Error ? error.message : "データベースクエリでエラーが発生しました";
      throw new Error(`クエリ実行エラー: ${errorMessage}`);
    }
  }

  private async handleGetTableSchema(args: any) {
    const parsed = GetTableSchemaSchema.parse(args);
    
    try {
      const schema = await this.dbManager.getTableSchema(parsed.tableName);
      
      if (schema.length === 0) {
        throw new Error(`テーブル '${parsed.tableName}' が見つかりません`);
      }

      const schemaInfo = {
        tableName: parsed.tableName,
        columns: schema.map((col: any) => ({
          name: col.name,
          type: col.type,
          nullable: !col.notnull,
          primaryKey: col.pk === 1,
          defaultValue: col.dflt_value
        }))
      };

      return {
        content: [
          {
            type: "text",
            text: `テーブル '${parsed.tableName}' のスキーマ情報:\n\n${JSON.stringify(schemaInfo, null, 2)}`
          } as TextContent
        ]
      };
    } catch (error) {
      const errorMessage = error instanceof Error ? error.message : "スキーマ取得でエラーが発生しました";
      throw new Error(`スキーマ取得エラー: ${errorMessage}`);
    }
  }

  private async handleListTables() {
    try {
      const tables = await this.dbManager.getTableNames();
      
      return {
        content: [
          {
            type: "text",
            text: `データベース内のテーブル一覧:\n\n${tables.map(table => `• ${table}`).join('\n')}`
          } as TextContent
        ]
      };
    } catch (error) {
      const errorMessage = error instanceof Error ? error.message : "テーブル一覧取得でエラーが発生しました";
      throw new Error(`テーブル一覧取得エラー: ${errorMessage}`);
    }
  }

  async start() {
    const transport = new StdioServerTransport();
    await this.server.connect(transport);
    console.error("SQLite MCP Server が開始されました");
  }

  async stop() {
    this.dbManager.close();
  }
}

// サーバー開始
async function main() {
  const dbPath = process.argv[2] || 'products.db';
  
  if (!require('fs').existsSync(dbPath)) {
    console.error(`エラー: データベースファイル '${dbPath}' が見つかりません`);
    process.exit(1);
  }

  const server = new SQLiteMCPServer(dbPath);
  
  // グレースフル・シャットダウン
  process.on('SIGINT', async () => {
    console.error('シャットダウン中...');
    await server.stop();
    process.exit(0);
  });

  await server.start();
}

// エラーハンドリング
process.on('unhandledRejection', (reason, promise) => {
  console.error('Unhandled Rejection at:', promise, 'reason:', reason);
  process.exit(1);
});

process.on('uncaughtException', (error) => {
  console.error('Uncaught Exception:', error);
  process.exit(1);
});

if (require.main === module) {
  main().catch((error) => {
    console.error("サーバー開始エラー:", error);
    process.exit(1);
  });
}

🔧 ステップ3：Claude Desktop設定

claude_desktop_config.jsonに以下の設定を追加：

{
  "mcpServers": {
    "sqlite-server": {
      "command": "npx",
      "args": ["ts-node", "/path/to/your/project/server.ts", "products.db"]
    }
  }
}

🚀 ステップ4：動作確認とテスト

サーバーの起動

npx ts-node server.ts products.db

Claude Desktopでのテスト例

以下のような質問でテストしてみましょう：

基本的なクエリ

"製品データベースの全テーブルを教えて"

データ分析

"カテゴリ別の製品数を教えて"

複雑なクエリ

"価格が100ドル以上の電子機器の一覧を、在庫数と一緒に表示して"

スキーマ確認

"productsテーブルのスキーマを教えて"

期待される動作フロー

テーブル一覧の取得:
Claudeがlist_tablesツールを実行してデータベース構造を理解
スキーマの確認:
get_table_schemaツールでテーブル構造を確認
クエリの生成と実行:
適切なSELECTクエリを生成し、query_databaseツールで実行
結果の解釈:
データベースの結果を分析してユーザーに回答

🛡️ セキュリティ機能の詳細

1. 読み取り専用の強制

データベースをOPEN_READONLYモードで開く
SELECT文のみを許可する厳密なバリデーション
INSERT、UPDATE、DELETE等の操作を完全にブロック

2. SQLインジェクション防止

許可されたキーワードのホワイトリスト
禁止されたキーワードのブラックリスト
複数SQL文の実行を防止

3. エラーハンドリング

詳細なエラーメッセージでデバッグを支援
セキュリティ情報の漏洩を防止
適切なログ出力

📊 実用的な応用例

ビジネスインテリジェンス

// 売上分析のためのクエリ例
const analyticsQueries = {
  monthlySales: `
    SELECT 
      strftime('%Y-%m', order_date) as month,
      COUNT(*) as order_count,
      SUM(p.price * o.quantity) as total_revenue
    FROM orders o 
    JOIN products p ON o.product_id = p.id
    GROUP BY month 
    ORDER BY month DESC
  `,
  
  topProducts: `
    SELECT 
      p.name,
      p.category,
      COUNT(o.id) as order_count,
      SUM(o.quantity) as total_sold
    FROM products p 
    LEFT JOIN orders o ON p.id = o.product_id
    GROUP BY p.id 
    ORDER BY order_count DESC
    LIMIT 10
  `
};

在庫管理

-- 在庫が少ない商品の確認
SELECT name, category, stock_quantity
FROM products 
WHERE stock_quantity < 30 
ORDER BY stock_quantity ASC;

-- カテゴリ別在庫状況
SELECT 
  category,
  COUNT(*) as product_count,
  AVG(stock_quantity) as avg_stock,
  MIN(stock_quantity) as min_stock
FROM products 
GROUP BY category;

🎯 ベストプラクティス

1. パフォーマンスの最適化

// インデックスの活用を促すクエリ設計
// データベース接続のプーリング
// クエリ結果のキャッシュ（必要に応じて）

2. エラー処理の改善

private handleDatabaseError(error: any): string {
  if (error.code === 'SQLITE_BUSY') {
    return 'データベースが使用中です。しばらく後に再試行してください。';
  } else if (error.code === 'SQLITE_LOCKED') {
    return 'データベースがロックされています。';
  } else {
    return `データベースエラー: ${error.message}`;
  }
}

3. 監査とログ

private logQuery(query: string, results: any[], executionTime: number) {
  console.log({
    timestamp: new Date().toISOString(),
    query: query.substring(0, 100) + (query.length > 100 ? '...' : ''),
    resultCount: results.length,
    executionTime: `${executionTime}ms`
  });
}

📈 スケーラビリティの考慮

より大規模なデータベースへの対応

PostgreSQL/MySQL対応: 同じパターンで他のデータベースにも対応可能
接続プーリング: 複数のクライアントからの同時アクセスに対応
クエリの最適化: EXPLAINを使った実行計画の分析
キャッシュ戦略: 頻繁にアクセスされるデータのキャッシュ

パフォーマンス監視

class PerformanceMonitor {
  private static queryStats = new Map<string, { count: number, totalTime: number }>();

  static recordQuery(query: string, executionTime: number) {
    const stats = this.queryStats.get(query) || { count: 0, totalTime: 0 };
    stats.count++;
    stats.totalTime += executionTime;
    this.queryStats.set(query, stats);
  }

  static getSlowQueries(threshold = 1000): Array<{ query: string, avgTime: number }> {
    const slow = [];
    for (const [query, stats] of this.queryStats) {
      const avgTime = stats.totalTime / stats.count;
      if (avgTime > threshold) {
        slow.push({ query, avgTime });
      }
    }
    return slow.sort((a, b) => b.avgTime - a.avgTime);
  }
}

🔍 トラブルシューティング

よくある問題と解決法

データベース接続エラー

# ファイルパスを確認
ls -la products.db
# パーミッションを確認
chmod 644 products.db

クエリが遅い場合

-- インデックスの確認
PRAGMA index_list('products');
-- クエリプランの確認  
EXPLAIN QUERY PLAN SELECT * FROM products WHERE category = 'Electronics';

メモリ使用量の問題

// 大きな結果セットの場合は LIMIT を使用
if (!query.toUpperCase().includes('LIMIT')) {
  query += ' LIMIT 1000';
}

📝 まとめ

今回は、SQLiteデータベースへの読み取り専用アクセスをMCP経由で安全に提供する方法を学びました。

重要なポイント

セキュリティファースト: 読み取り専用の徹底とSQLインジェクション防止
構造化されたアプローチ: スキーマ情報の提供とクエリバリデーション
エラーハンドリング: 適切なエラー処理と情報提供
スケーラビリティ: より大規模なシステムへの対応を考慮した設計

応用可能性

このパターンは以下の用途に応用できます：

ビジネスインテリジェンス: 売上分析、KPI監視
在庫管理: 商品管理、発注支援
顧客管理: 顧客データの分析と検索
レポート生成: 定期レポートの自動生成

次回は、外部Web API統合MCPをテーマに、RESTful APIを安全にラップし、LLMから活用する実装パターンを解説します。お楽しみに！

🔗 参考リンク