[特殊字符] MySQL MCP 開發實戰：打造智能數據庫操作助手

💡 簡介：本文詳細介紹如何利用MCP（Model-Control-Panel）框架開發MySQL數據庫操作工具，使AI助手能夠直接執行數據庫操作。

📚 目錄

引言
MCP框架簡介
項目架構設計
開發環境搭建
核心代碼實現
錯誤處理策略
運行和部署
使用示例
項目擴展與優化
總結
參考資料和項目源碼

🌟 引言

在現代軟件開發中，數據庫操作是不可或缺的一部分。隨著人工智能技術的發展，將AI與數據庫操作工具結合起來成為一種新趨勢。本文將介紹如何利用MCP（Model-Control-Panel）框架開發一個MySQL數據庫操作工具，使AI助手能夠直接執行數據庫操作。

🔍 MCP框架簡介

MCP（Model-Control-Panel）是一個創新的工具框架，它允許我們將工具函數暴露為API，使模型（如AI助手）能夠直接調用這些函數。通過MCP，我們可以將繁瑣的數據庫操作封裝成簡單的函數調用，大大提高開發效率。

🏗? 項目架構設計

MySQL MCP工具的核心是一個Python腳本，它使用FastMCP服務器暴露MySQL操作函數。整個項目架構如下：

配置管理：支持命令行參數、環境變量和默認配置
連接管理：處理數據庫連接、重試和錯誤報告
工具函數：封裝MySQL操作為易用的API
錯誤處理：提供詳細的錯誤信息和原因分析

🛠? 開發環境搭建

首先，我們需要準備開發環境：

安裝Python 3.12或更高版本
安裝所需依賴：
- mcp[cli] >= 1.5.0
- mysql-connector-python >= 9.2.0

項目的pyproject.toml文件定義了這些依賴：

[project]
name = "mysql-mcp"
version = "0.1.0"
description = "MySQL MCP 工具"
readme = "README.md"
requires-python = ">=3.12"
dependencies = ["mcp[cli]>=1.5.0","mysql-connector-python>=9.2.0",
]

💻 核心代碼實現

初始化MCP服務器

首先，我們導入必要的模塊并初始化FastMCP服務器：

from typing import Any, List, Dict, Optional
import os
import argparse
import mysql.connector
from mysql.connector import Error
from mcp.server.fastmcp import FastMCP# 初始化 FastMCP server
mcp = FastMCP("mysql")

配置管理

為了使工具更加靈活，我們實現了多層次的配置系統：

# 數據庫連接配置默認值
DEFAULT_DB_CONFIG = {"host": os.getenv("MYSQL_HOST", "localhost"),"port": int(os.getenv("MYSQL_PORT", "3306")),"user": os.getenv("MYSQL_USER", "root"),"password": os.getenv("MYSQL_PASSWORD", "root"),"database": os.getenv("MYSQL_DATABASE", ""),"connection_timeout": int(os.getenv("MYSQL_CONNECTION_TIMEOUT", "10")),"connect_retry_count": int(os.getenv("MYSQL_CONNECT_RETRY_COUNT", "3"))
}

命令行參數解析：

def parse_args():parser = argparse.ArgumentParser(description='MySQL MCP服務')parser.add_argument('--host', type=str, help='數據庫主機地址')parser.add_argument('--port', type=int, help='數據庫端口')parser.add_argument('--user', type=str, help='數據庫用戶名')parser.add_argument('--password', type=str, help='數據庫密碼')parser.add_argument('--database', type=str, help='數據庫名稱')parser.add_argument('--connection-timeout', type=int, help='連接超時時間(秒)')parser.add_argument('--connect-retry-count', type=int, help='連接重試次數')return parser.parse_args()

數據庫連接管理

數據庫連接是工具的核心部分，我們實現了連接重試和詳細的錯誤報告：

def get_connection(db_config=None):"""獲取數據庫連接Args:db_config: 數據庫連接配置參數，如果為None則使用默認配置Returns:數據庫連接對象"""# 配置處理邏輯...retry_count = 0last_error = Nonemax_retries = db_config.get("connect_retry_count", 3)while retry_count < max_retries:try:# 創建連接...return connexcept Error as e:last_error = eretry_count += 1# 重試邏輯...# 詳細錯誤報告error_message = f"數據庫連接錯誤(重試 {retry_count} 次后): {last_error}"if "Can't connect to MySQL server" in str(last_error):error_message += f"\n無法連接到MySQL服務器，請檢查主機 {db_config['host']} 和端口 {db_config['port']} 是否正確"# 更多詳細信息...raise Exception(error_message)

工具函數實現

下面是幾個關鍵工具函數的實現：

1?? 執行SQL查詢

@mcp.tool()
async def execute_query(query: str, params: Optional[List[Any]] = None, db_config: Optional[Dict[str, Any]] = None) -> Dict[str, Any]:"""執行SQL查詢語句，返回查詢結果"""try:conn = get_connection(db_config)cursor = conn.cursor(dictionary=True)cursor.execute(query, params)# 判斷查詢類型并返回適當的結果query_upper = query.strip().upper()if query_upper.startswith("SELECT") or query_upper.startswith("SHOW") or query_upper.startswith("DESCRIBE"):results = cursor.fetchall()return {"success": True,"rows": results,"row_count": len(results)}else:# 非查詢操作（如INSERT, UPDATE, DELETE）conn.commit()return {"success": True,"affected_rows": cursor.rowcount,"last_insert_id": cursor.lastrowid}except Error as e:# 錯誤處理和詳細分析error_message = f"執行查詢失敗: {str(e)}"if "Unknown column" in str(e):error_message += "\n原因：查詢中包含未知的列名"# 更多錯誤分析...return {"error": error_message, "query": query}finally:# 確保資源釋放if 'conn' in locals() and conn.is_connected():cursor.close()conn.close()

2?? 列出表

@mcp.tool()
async def list_tables(database_name: Optional[str] = None, db_config: Optional[Dict[str, Any]] = None) -> Dict[str, Any]:"""列出指定數據庫中的所有表"""try:conn = get_connection(db_config)cursor = conn.cursor()# 執行適當的查詢if database_name:cursor.execute(f"SHOW TABLES FROM {database_name}")else:cursor.execute("SHOW TABLES")tables = [table[0] for table in cursor.fetchall()]return {"success": True,"database": database_name or conn.database,"tables": tables,"count": len(tables)}except Error as e:# 錯誤處理...

3?? 獲取表結構

@mcp.tool()
async def describe_table(table_name: str, db_config: Optional[Dict[str, Any]] = None) -> Dict[str, Any]:"""獲取表結構"""try:conn = get_connection(db_config)cursor = conn.cursor(dictionary=True)cursor.execute(f"DESCRIBE {table_name}")columns = cursor.fetchall()return {"success": True,"table": table_name,"columns": columns}except Error as e:# 錯誤處理...

4?? 數據操作函數

此外，我們還實現了一系列數據操作函數：

create_table: 創建新表
insert_data: 插入數據
update_data: 更新數據
delete_data: 刪除數據
use_database: 切換數據庫

?? 錯誤處理策略

MySQL MCP工具的一大特色是提供了詳細的錯誤分析和報告。對于每種常見的數據庫錯誤，我們都提供了簡潔明了的解釋和可能的解決方案：

# 示例：插入數據時的錯誤處理
error_message = f"插入數據失敗: {str(e)}"
if "doesn't exist" in str(e):error_message += f"\n原因：表 {table_name} 不存在"
elif "Unknown column" in str(e):error_message += "\n原因：插入數據中包含表中不存在的列"
elif "cannot be null" in str(e):error_message += "\n原因：某個NOT NULL字段被設置為NULL值"
elif "Duplicate entry" in str(e):error_message += "\n原因：插入的數據違反了唯一鍵約束"
elif "Data too long" in str(e):error_message += "\n原因：插入的數據超出了字段的長度限制"

🚀 運行和部署

最后，我們設置了入口點并啟動MCP服務器：

if __name__ == "__main__":# 從命令行參數獲取配置GLOBAL_DB_CONFIG = get_config_from_args()# 啟動MCP服務器mcp.run(transport='stdio')

配置和啟動MCP服務的方式有多種：

直接運行腳本：

python mysql-mcp.py --host localhost --port 3306 --user root --password your_password --database your_database

通過環境變量：

export MYSQL_HOST=localhost
export MYSQL_PORT=3306
export MYSQL_USER=root
export MYSQL_PASSWORD=your_password
export MYSQL_DATABASE=your_database
python mysql-mcp.py

通過Cursor IDE配置：
在~/.cursor/mcp.json中添加配置：

{"mcpServers": {"mysql-mcp": {"command": "/path/to/uv","args": ["--directory","/path/to/mysql-mcp","run","mysql-mcp.py","--host", "xxx.xxx.xxx.xxx","--port", "3306","--user", "root","--password", "********","--database", "your_database"]}}
}

📊 使用示例

配置完成后，在Cursor IDE中，AI助手可以直接調用MySQL MCP工具：

# 查詢所有數據庫# 列出當前數據庫的所有表# 查詢用戶表中年齡大于18的用戶# 創建新表# 插入數據# 更新數據# 刪除數據

🔧 項目擴展與優化

MySQL MCP工具還有許多可擴展之處：

事務支持：添加事務控制函數，如begin_transaction、commit和rollback
批量操作：支持批量插入、更新和刪除
查詢構建器：提供SQL查詢構建助手，簡化復雜查詢的構造
讀寫分離：支持主從數據庫配置
連接池：實現連接池管理，提高性能
安全增強：添加輸入驗證和SQL注入防護

📝 總結

通過MySQL MCP工具，我們成功將復雜的數據庫操作封裝為簡單直觀的API，使AI助手能夠直接執行數據庫任務。該工具的主要優勢包括：

簡單易用：清晰的API設計，易于理解和使用
錯誤處理：詳細的錯誤信息和原因分析
靈活配置：支持多種配置方式
安全可靠：參數化查詢防止SQL注入
完整功能：涵蓋常見的數據庫操作

MySQL MCP工具為開發者提供了一種全新的數據庫交互方式，特別適合與AI工具集成，大大簡化了數據庫操作流程，提高了開發效率。

希望本文能幫助你了解MCP框架的強大功能，并啟發你開發更多創新的工具應用。

📚 參考資料

MySQL Connector/Python 文檔
！！！項目源碼