💡 简介：本文详细介绍如何利用MCP（Model-Control-Panel）框架开发MySQL数据库操作工具，使AI助手能够直接执行数据库操作。
📚 目录
🌟 引言
在现代软件开发中，数据库操作是不可或缺的一部分。随着人工智能技术的发展，将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 = 0
last_error = None
max_retries = db_config.get("connect_retry_count", 3)
while retry_count < max_retries:
try:
# 创建连接...
return conn
except Error as e:
last_error = e
retry_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框架的强大功能，并启发你开发更多创新的工具应用。