MCP 資料庫整合

Model Context Protocol（MCP）入門

James Chapman

AI Curriculum Manager, DataCamp

為什麼要在 Timezone Server 加上資料庫？

查詢能力：不需載入整個資料集即可篩選、搜尋、排序
可擴充性：用資料庫 indexes 加速查詢

為什麼要在 Timezone Server 加上資料庫？

查詢能力：不需載入整個資料集即可篩選、搜尋、排序
可擴充性：用資料庫 indexes 加速查詢
並行性：多個工具呼叫或用戶端可安全存取 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
- Prompt injection → 透過提示惡意注入程式碼

SELECT timezone FROM locations WHERE timezone LIKE ? LIMIT 50

MCP 伺服器宜採唯讀且 範圍嚴謹 的資料庫存取
設限：最大列數與逾時

一起來練習吧！

Model Context Protocol（MCP）入門