[特殊字符] 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 = 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框架的强大功能，并启发你开发更多创新的工具应用。

📚 参考资料

MySQL Connector/Python 文档
！！！项目源码