🎯 SQL MCP 伺服器：讓AI輕鬆讀懂你的資料庫 🌟 這是什麼神奇的東西？ 想像一下，你有一個聰明的助手，它不僅能和你聊天，還能直接幫你查看資料庫裡的資料！這就是 SQL MCP 伺服器要做的事情。 簡單來說： 它就像是AI和你的SQL資料庫之間的翻譯官，讓Claude這樣的AI助手能夠： 🔍 查看你的資料庫結構（就像翻開一本書的目錄） 📊 執行查詢來獲取資料（就像問「給我看看銷售資料」） 💬 用自然語言和你討論資料內容 ⚠️ 小提醒： 這個工具還在成長中，就像一個聰明的孩子，功能會越來越強大！ 🛠️ 它能做什麼？ 1. 📖 read_query - 資料查詢大師 作用： 幫你執行 SELECT 查詢 就像： 你問「今年銷售額多少？」，它幫你去資料庫裡找答案 輸入： 一個SQL查詢語句 輸出： 整理好的資料表格 安全保障： 只能讀取資料，不能修改（就像只給了閱讀權限的圖書館卡） 2. 🏗️ get_schema - 資料庫結構探索者 作用： 展示你的資料庫長什麼樣 就像： 給你一張房子的設計圖，告訴你哪個房間是什麼 輸入： 不需要任何參數 輸出： 完整的資料庫結構說明書 🚀 開始使用吧！ 方法一：用 uv（推薦，最簡單！） # 不需要安裝，直接執行！ uvx mcp-sql-server \  --db-host 你的資料庫地址 \  --db-user 使用者名稱 \  --db-password 密碼 \  --db-database 資料庫名稱 方法二：用傳統的 pip # 先安裝 pip install mcp-sql-server # 然後執行 python -m mcp_sql_server \  --db-host 你的資料庫地址 \  --db-user 使用者名稱 \  --db-password 密碼 \  --db-database 資料庫名稱 🔧 配置你的AI助手 讓Claude Desktop認識這個工具 打開你的 claude_desktop_config.json 檔案，加入這段神奇的咒語： {  "mcpServers": {    "sql": {      "command": "uvx",      "args": [        "mcp-sql-server",        "--db-host", "你的資料庫地址",        "--db-user", "使用者名稱",        "--db-password", "密碼",        "--db-database", "資料庫名稱"      ]    }  } } 💡 小貼士： 記得把「你的資料庫地址」這些替換成真實的資訊喔！ 其他配置方式 🐳 Docker 方式（適合喜歡容器的朋友） {  "mcpServers": {    "sql": {      "command": "docker",      "args": [        "run", "--rm", "-i",        "--mount", "type=bind,src=/你的路徑,dst=/你的路徑",        "mcp/sql",        "--db-host", "你的資料庫地址",        "--db-user", "使用者名稱",        "--db-password", "密碼",        "--db-database", "資料庫名稱"      ]    }  } } 🐍 Python 直接執行 {  "mcpServers": {    "sql": {      "command": "python",      "args": [        "-m", "mcp_sql_server",        "--db-host", "你的資料庫地址",        "--db-user", "使用者名稱",        "--db-password", "密碼",        "--db-database", "資料庫名稱"      ]    }  } } 在Zed編輯器中使用 如果你用Zed，在 settings.json 裡加上： {  "context_servers": {    "mcp-sql-server": {      "command": {        "path": "uvx",        "args": [          "mcp-sql-server",          "--db-host", "你的資料庫地址",          "--db-user", "使用者名稱",          "--db-password", "密碼",          "--db-database", "資料庫名稱"        ]      }    }  } } 🔍 遇到問題？來調試一下！ 使用MCP檢查器 npx @modelcontextprotocol/inspector uvx mcp-sql-server \  --db-host 你的資料庫地址 \  --db-user 使用者名稱 \  --db-password 密碼 \  --db-database 資料庫名稱 查看日誌檔案 tail -n 20 -f ~/Library/Logs/Claude/mcp*.log 這就像是看伺服器的「日記」，能告訴你發生了什麼！ 🧪 開發者模式 如果你想自己改造這個工具： 測試你的修改 用檢查器測試： 執行上面的調試命令 用Claude測試： 修改配置檔案指向你的開發版本 開發配置範例 Docker開發版 {  "mcpServers": {    "sql": {      "command": "docker",      "args": [        "run", "--rm", "-i",        "--mount", "type=bind,src=/你的專案路徑,dst=/projects",        "mcp/sql",        "--db-host", "你的資料庫地址",        "--db-user", "使用者名稱",        "--db-password", "密碼",        "--db-database", "資料庫名稱"      ]    }  } } UV開發版 {  "mcpServers": {    "sql": {      "command": "uv",      "args": [        "--directory", "/你的專案路徑/mcp-servers/src/sql",        "run", "mcp-sql-server",        "--db-host", "你的資料庫地址",        "--db-user", "使用者名稱",        "--db-password", "密碼",        "--db-database", "資料庫名稱"      ]    }  } } 🏗️ 建構Docker映像檔 想自己打包？很簡單： cd src/sql docker build -t mcp/sql . 🎉 開始你的資料探索之旅！ 現在你已經有了這個強大的工具，可以讓AI助手幫你： 📈 分析銷售趨勢 👥 查看使用者資料 📊 產生報表 🔍 快速查找資訊 記住，這就像給你的AI助手配了一副「資料眼鏡」，讓它能看懂你的資料庫。現在就開始你的資料探索之旅吧！ 💡 最後提醒： 別忘了保護好你的資料庫憑證，不要把密碼洩露給別人喔！ 下面是主要的 MCP SQL reader 的服務 code: import pymysql import json import psycopg2 import logging from contextlib import closing from mcp.server.models import InitializationOptions import mcp.types as types from mcp.server import NotificationOptions, Server import mcp.server.stdio from typing import Any, Literal # 設定日誌記錄器，用於記錄服務器運作狀況 logger = logging.getLogger('mcp_sql_server') logger.info("Starting MCP SQL Server") class SqlReadOnlyServer: """ SQL 唯讀服務器類別 這個類別提供與 MySQL 或 PostgreSQL 資料庫交互的功能， 專門設計為只能執行讀取操作，確保資料安全性。 主要功能： - 執行 SELECT 查詢 - 獲取資料庫架構資訊 - 支援 MySQL 和 PostgreSQL 兩種資料庫 """ def __init__(self, host: str, user: str, password: str, database: str, db_type: Literal["mysql", "postgres"] = "mysql", port: str = None): """ 初始化 SQL 唯讀服務器 建立資料庫連線所需的基本參數設定 參數說明: host (str): 資料庫主機地址（例如：localhost 或 IP 地址） user (str): 資料庫使用者名稱 password (str): 資料庫密碼 database (str): 要連接的資料庫名稱 db_type (str): 資料庫類型（"mysql" 或 "postgres"），預設為 "mysql" port (str): 資料庫連接埠（PostgreSQL 必須提供） 例外處理: ValueError: 如果選擇 PostgreSQL 但未提供連接埠會拋出錯誤 """ self.host = host self.user = user self.password = password self.database = database self.db_type = db_type self.port = port # PostgreSQL 必須指定連接埠，否則無法連接 if db_type == "postgres" and not port: raise ValueError("Port is required for PostgreSQL connection") def _get_schema_for_llm(self) -> str: """ 獲取資料庫架構資訊，格式化為適合 LLM 理解的 JSON 格式 這個方法會掃描整個資料庫，收集所有表格的欄位資訊， 包括欄位名稱、資料類型、是否可為空值等詳細資訊。 返回值: str: JSON 格式的資料庫架構字串，包含所有表格和欄位的詳細資訊 架構資訊包含: - 表格名稱 - 欄位名稱和資料類型 - 是否允許 NULL 值 - 預設值 - 主鍵/索引資訊 - 其他欄位屬性 """ if self.db_type == "mysql": # === MySQL 資料庫架構查詢 === connection = pymysql.connect( host=self.host, user=self.user, password=self.password, database=self.database ) schema = {} # 用於儲存所有表格架構的字典 try: with connection.cursor() as cursor: # 查詢 MySQL 的 INFORMATION_SCHEMA 來獲取欄位資訊 # 這個查詢會返回指定資料庫中所有表格的欄位詳細資訊 query = """ SELECT TABLE_NAME, -- 表格名稱 COLUMN_NAME, -- 欄位名稱 DATA_TYPE, -- 基本資料類型（如 varchar, int） COLUMN_TYPE, -- 完整欄位類型（如 varchar(255)） IS_NULLABLE, -- 是否允許 NULL COLUMN_DEFAULT, -- 預設值 COLUMN_KEY, -- 鍵類型（PRI=主鍵, UNI=唯一鍵等） EXTRA -- 額外屬性（如 auto_increment） FROM INFORMATION_SCHEMA.COLUMNS WHERE TABLE_SCHEMA = %s -- 只查詢指定的資料庫 ORDER BY TABLE_NAME, ORDINAL_POSITION; -- 按表格名稱和欄位順序排序 """ cursor.execute(query, (self.database,)) results = cursor.fetchall() # 將查詢結果組織成結構化的字典格式 for row in results: table_name = row[0] column_info = { "name": row[1], # 欄位名稱 "data_type": row[2], # 資料類型 "column_type": row[3], # 完整類型定義 "is_nullable": row[4], # 是否可為空 "default": row[5], # 預設值 "key": row[6], # 鍵類型 "extra": row[7] # 額外屬性 } # 如果表格還沒有在架構字典中，建立新的列表 if table_name not in schema: schema[table_name] = [] schema[table_name].append(column_info) finally: # 確保連線被正確關閉，避免連線洩漏 connection.close() else: # PostgreSQL 資料庫架構查詢 # === PostgreSQL 資料庫架構查詢 === connection = psycopg2.connect( database=self.database, user=self.user, password=self.password, host=self.host, port=self.port ) schema = {} # 用於儲存所有表格架構的字典 try: with connection.cursor() as cursor: # 查詢 PostgreSQL 的 information_schema 來獲取欄位資訊 # 只查詢 public schema 中的表格（PostgreSQL 的預設 schema） query = """ SELECT table_name, -- 表格名稱 column_name, -- 欄位名稱 data_type, -- 資料類型 is_nullable, -- 是否允許 NULL column_default, -- 預設值 character_maximum_length -- 字元長度限制（適用於文字類型） FROM information_schema.columns WHERE table_schema = 'public' -- 只查詢 public schema ORDER BY table_name, ordinal_position; -- 按表格名稱和欄位順序排序 """ cursor.execute(query) results = cursor.fetchall() # 將查詢結果組織成結構化的字典格式 for row in results: table_name = row[0] column_info = { "name": row[1], # 欄位名稱 "data_type": row[2], # 資料類型 "is_nullable": row[3], # 是否可為空 "default": row[4], # 預設值 "max_length": row[5] # 最大長度 } # 如果表格還沒有在架構字典中，建立新的列表 if table_name not in schema: schema[table_name] = [] schema[table_name].append(column_info) finally: # 確保連線被正確關閉 connection.close() # 將架構字典轉換為格式化的 JSON 字串，方便 LLM 理解 return json.dumps(schema, indent=2) def _execute_query(self, query: str) -> list[dict]: """ 執行 SQL 查詢並返回結果 這個方法負責實際執行 SQL 查詢語句，並將結果格式化為 字典列表的形式，每個字典代表一行資料。 參數: query (str): 要執行的 SQL 查詢語句（必須是 SELECT 語句） 返回值: list[dict]: 查詢結果列表，每個字典包含一行資料，鍵為欄位名稱 例外處理: ValueError: 如果查詢執行過程中發生錯誤 安全性說明: 這個方法只用於執行查詢，不會進行任何資料修改操作 """ if self.db_type == "mysql": # === MySQL 查詢執行 === connection = pymysql.connect( host=self.host, user=self.user, password=self.password, database=self.database ) try: # 使用 DictCursor 讓結果自動轉換為字典格式 # 這樣每一行資料都會是 {欄位名: 值} 的字典形式 with connection.cursor(pymysql.cursors.DictCursor) as cursor: cursor.execute(query) # 執行 SQL 查詢 results = cursor.fetchall() # 獲取所有查詢結果 return results finally: # 確保連線被關閉，釋放資源 connection.close() else: # PostgreSQL 查詢執行 # === PostgreSQL 查詢執行 === connection = psycopg2.connect( database=self.database, user=self.user, password=self.password, host=self.host, port=self.port ) try: with connection.cursor() as cursor: cursor.execute(query) # 執行 SQL 查詢 # PostgreSQL 的 cursor 不像 MySQL 有 DictCursor # 所以需要手動將結果轉換為字典格式 columns = [desc[0] for desc in cursor.description] # 獲取欄位名稱 results = [dict(zip(columns, row)) for row in cursor.fetchall()] # 組合成字典 return results finally: # 確保連線被關閉 connection.close() async def main(host: str, user: str, password: str, database: str, db_type: str = "postgres", port: str = '5432'): """ MCP SQL 服務器的主要啟動函數 這個函數負責初始化並啟動整個 MCP 服務器，包括： 1. 建立資料庫連線物件 2. 設定 MCP 服務器實例 3. 註冊可用的工具（tools） 4. 處理工具呼叫請求 5. 啟動服務器並開始監聽 參數: host (str): 資料庫主機地址 user (str): 資料庫使用者名稱 password (str): 資料庫密碼 database (str): 資料庫名稱 db_type (str): 資料庫類型，預設為 "postgres" port (str): 資料庫連接埠，預設為 '5432' """ # 建立資料庫連線物件 db = SqlReadOnlyServer(host=host, user=user, password=password, database=database, db_type=db_type, port=port) # 建立 MCP 服務器實例 server = Server("mcp-sql-server") @server.list_tools() async def handle_list_tools() -> list[types.Tool]: """ 註冊並返回服務器提供的所有工具 這個函數定義了 AI 助手可以使用的工具清單。 每個工具都有名稱、描述和輸入參數的定義。 返回值: list[types.Tool]: 可用工具的清單 """ return [ # 工具 1: 執行 SELECT 查詢 types.Tool( name="read_query", description="在 SQL 資料庫上執行 SELECT 查詢，用於讀取和分析資料", inputSchema={ "type": "object", "properties": { "query": { "type": "string", "description": "要執行的 SELECT SQL 查詢語句，只允許讀取操作" }, }, "required": ["query"], # query 參數是必需的 }, ), # 工具 2: 獲取資料庫架構 types.Tool( name="get_schema", description="獲取資料庫的架構資訊，包括所有表格、欄位和資料類型", inputSchema={ "type": "object", "properties": {}, # 此工具不需要任何輸入參數 }, ), ] @server.call_tool() async def handle_call_tool( name: str, arguments: dict[str, Any] | None ) -> list[types.TextContent | types.ImageContent | types.EmbeddedResource]: """ 處理工具呼叫請求的核心函數 當 AI 助手想要使用某個工具時，這個函數會被呼叫。 它根據工具名稱執行相應的操作並返回結果。 參數: name (str): 要呼叫的工具名稱 arguments (dict): 工具的輸入參數（可能為 None） 返回值: list: 包含工具執行結果的內容清單 安全性檢查: - 確保 read_query 只能執行 SELECT 語句 - 所有例外都會被捕捉並返回錯誤訊息 """ try: if name == "get_schema": # === 處理獲取架構的請求 === # 呼叫資料庫物件的架構獲取方法 results = db._get_schema_for_llm() # 將結果包裝為文字內容返回 return [types.TextContent(type="text", text=str(results))] # 檢查是否有提供必要的參數 if not arguments: raise ValueError("Missing arguments") if name == "read_query": # === 處理查詢請求 === # 安全性檢查：確保只能執行 SELECT 查詢 query = arguments["query"].strip().upper() if not query.startswith("SELECT"): raise ValueError("只允許執行 SELECT 查詢，以確保資料安全性") # 執行查詢並獲取結果 results = db._execute_query(arguments["query"]) # 將結果包裝為文字內容返回 return [types.TextContent(type="text", text=str(results))] else: # 未知的工具名稱 raise ValueError(f"未知的工具: {name}") except Exception as e: # === 例外處理 === # 如果任何步驟發生錯誤，返回錯誤訊息而不是讓程式崩潰 # 這確保了服務器的穩定性 return [types.TextContent(type="text", text=f"錯誤: {str(e)}")] # === 啟動服務器 === async with mcp.server.stdio.stdio_server() as (read_stream, write_stream): logger.info("服務器正在使用 stdio 傳輸協定運行") # 執行服務器主迴圈 # 這會讓服務器開始監聽來自 AI 助手的請求 await server.run( read_stream, # 輸入串流（接收請求） write_stream, # 輸出串流（發送回應） InitializationOptions( server_name="sql", # 服務器名稱 server_version="0.1.0", # 版本號 capabilities=server.get_capabilities( notification_options=NotificationOptions(), # 通知選項 experimental_capabilities={}, # 實驗性功能 ), ), ) 🎯 註解重點： 1. 類別層級註解 SqlReadOnlyServer 類別的完整說明 每個方法的用途和工作原理 2. 函數參數詳解 每個參數的意義和用法 預設值的說明 必需參數的標註 3. 安全性說明 為什麼只允許 SELECT 查詢 連線管理和資源釋放 例外處理機制 4. 資料庫支援 MySQL 和 PostgreSQL 的不同處理方式 各自的查詢語法差異 連線參數需求 5. MCP 協定說明 工具註冊機制 請求處理流程 回應格式規範 6. 實作細節 為什麼使用 DictCursor JSON 格式化的目的 錯誤處理策略