[特殊字符] 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操作函数。整个项目架构如下：

1. 配置管理：支持命令行参数、环境变量和默认配置
2. 连接管理：处理数据库连接、重试和错误报告
3. 工具函数：封装MySQL操作为易用的API
4. 错误处理：提供详细的错误信息和原因分析

🛠️ 开发环境搭建

首先，我们需要准备开发环境：

1. 安装Python 3.12或更高版本
2. 安装所需依赖：
   • 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服务的方式有多种：

1. 直接运行脚本：

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

2. 通过环境变量：

   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
   

3. 通过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工具还有许多可扩展之处：

1. 事务支持：添加事务控制函数，如begin_transaction、commit和rollback
2. 批量操作：支持批量插入、更新和删除
3. 查询构建器：提供SQL查询构建助手，简化复杂查询的构造
4. 读写分离：支持主从数据库配置
5. 连接池：实现连接池管理，提高性能
6. 安全增强：添加输入验证和SQL注入防护

📝 总结

通过MySQL MCP工具，我们成功将复杂的数据库操作封装为简单直观的API，使AI助手能够直接执行数据库任务。该工具的主要优势包括：

1. 简单易用：清晰的API设计，易于理解和使用
2. 错误处理：详细的错误信息和原因分析
3. 灵活配置：支持多种配置方式
4. 安全可靠：参数化查询防止SQL注入
5. 完整功能：涵盖常见的数据库操作

MySQL MCP工具为开发者提供了一种全新的数据库交互方式，特别适合与AI工具集成，大大简化了数据库操作流程，提高了开发效率。

希望本文能帮助你了解MCP框架的强大功能，并启发你开发更多创新的工具应用。

📚 参考资料

1. MySQL Connector/Python 文档
2. ！！！项目源码

---