MCPデータベース統合

Model Context Protocol（MCP）入門

James Chapman

AI Curriculum Manager, DataCamp

タイムゾーンサーバーにデータベースを追加する理由

クエリの利便性: データセット全体を読み込まずにフィルタリング・検索・ソートが可能
スケーラビリティ: データベースインデックスによる高速検索

データベース接続のアイコン

タイムゾーンサーバーにデータベースを追加する理由

クエリの利便性: データセット全体を読み込まずにフィルタリング・検索・ソートが可能
スケーラビリティ: データベースインデックスによる高速検索
同時実行性: 複数のツール呼び出しやクライアントが安全にDBへアクセス可能
本番対応: バックアップ、レプリケーション、単一の信頼できるデータソース

データベースのメリット図

データアクセス: ツール vs. リソース

ツールアイコン

動的、ユーザー入力依存、書き込み操作
例: データ分析・操作

リソースアイコン

読み取り専用、静的データ、参照情報
例: ガイドライン、ドキュメント、ポリシー

接続のライフサイクル

データベース接続の図

データベースリソース: timezone_server.py

from mcp.server.fastmcp import FastMCP
import sqlite3

mcp = FastMCP("Timezone Converter")

# Create connection at startup; reuse in handlers
conn = sqlite3.connect("timezones.db")
conn.row_factory = sqlite3.Row

@mcp.resource("db://timezones")
def get_locations() -> str:
    try:
        cursor = conn.execute("SELECT timezone FROM locations")
        rows = cursor.fetchall()
        return "\n".join(r["timezone"] for r in rows)
    except sqlite3.Error as e: return f"Error: {e}"

データベース検索ツール: timezone_server.py

@mcp.tool()
def lookup_locations(prefix: str) -> str:
    """Find timezones that contain the given prefix."""
    try:
        cursor = conn.execute(
            "SELECT timezone FROM locations WHERE timezone LIKE ? LIMIT 50", (f"%{prefix}%",))
        rows = cursor.fetchall()
        return "\n".join(r["timezone"] for r in rows)
    except sqlite3.Error as e: return f"Error: {e}"


if __name__ == "__main__":
    try:
        mcp.run(transport="stdio")
    finally:
        conn.close()

安全性と本番環境での注意事項

パラメーター化クエリ（?）: ユーザーやLLMの入力をSQLに直接埋め込まない
- プロンプトインジェクション → プロンプトを通じた悪意あるコード挿入

SELECT timezone FROM locations WHERE timezone LIKE ? LIMIT 50

MCPサーバーには読み取り専用かつ最小限のデータベースアクセスを推奨
制限を設ける: 最大行数とタイムアウト

練習しましょう！

Model Context Protocol（MCP）入門