[特殊字符] 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. ！！！项目源码

本文来自互联网用户投稿，该文观点仅代表作者本人，不代表本站立场。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如若转载，请注明出处：http://www.mzph.cn/bicheng/77158.shtml

如若内容造成侵权/违法违规/事实不符，请联系多彩编程网进行投诉反馈email:809451989@qq.com，一经查实，立即删除！

相关文章

Dify部署过程中的错误和解决方案汇总

Dify部署过程中的错误和解决方案汇总

本文仅限于记录Dify部署及使用过程中的BUG和解决方案 1. Dify配置SearXNG时报错&#xff1a; 报错内容&#xff1a; PluginInvokeError: {"args":{},"error_type":"ToolProviderCredentialValidationError","message":"Error 4…
阅读更多...
C#中async await异步关键字用法和异步的底层原理

C#中async await异步关键字用法和异步的底层原理

目录 C#异步编程一、异步编程基础二、异步方法的工作原理三、代码示例四、编译后的底层实现五、总结 C#异步编程 一、异步编程基础 异步编程是啥玩意儿 就是让程序在干等着某些耗时操作&#xff08;比如等网络响应、读写文件啥的&#xff09;的时候&#xff0c;能把线程腾出来…
阅读更多...
安全教育知识竞赛答题小程序怎么做

安全教育知识竞赛答题小程序怎么做

以下是制作安全教育知识竞赛答题小程序的一般步骤&#xff1a; 一、准备阶段 注册小程序账号&#xff1a;前往微信公众平台&#xff0c;注册一个小程序账号&#xff0c;主体类型可根据实际情况选择个人或企业等&#xff0c;注册成功后登录获取appid。 下载安装开发工具&#x…
阅读更多...
记录待办事项的便签软件有没有推荐的？

记录待办事项的便签软件有没有推荐的？

在快节奏的现代生活中&#xff0c;我们每天都要处理大量的工作任务和生活琐事&#xff0c;稍有不慎就可能遗漏重要事项。你是否经常遇到这样的情况&#xff1a;明明记得有件事要做&#xff0c;却怎么也想不起来是什么&#xff1b;或者手头同时有好几项任务&#xff0c;却不知道…
阅读更多...
实验四 中断实验

实验四 中断实验

一、实验目的 掌握中断服务程序的编写。 二、实验电路 三、实验内容 1&#xff0e;实验用PC机内部的中断控制器8259A&#xff0c;中断源用TPC-ZK实验箱上的单脉冲电路&#xff0c;将单脉冲电路的输出接中断请求信号IRQ&#xff0c;每按一次单脉冲按键产生一次…
阅读更多...
React 项目src文件结构

React 项目src文件结构

SCSS 组件库 SCSS为预处理器 支持除原生CSS外的其他语句 别名路径 在项目下的第一级目录就加入craco.config.js文件并且修改packpage.js 中的部分 // 扩展webpage的配置const path require(path)module.exports {// exports配置webpack:{// 配置别名alias:{:path.resolve(__d…
阅读更多...
Cursor入门教程-JetBrains过度向

Cursor入门教程-JetBrains过度向

Cursor使用笔记 **前置&#xff1a;**之前博主使用的是JetBrains的IDE&#xff0c;VSCode使用比较少&#xff0c;所以会尽量朝着JetBrains的使用习惯及样式去调整。 一、设置语言为中文 如果刚上手Cursor&#xff0c;那么肯定对Cursor中的众多选项配置项不熟悉&#xff0c;这…
阅读更多...
Linux上位机开发实践（SoC和MCU的差异）

Linux上位机开发实践（SoC和MCU的差异）

【 声明&#xff1a;版权所有&#xff0c;欢迎转载&#xff0c;请勿用于商业用途。 联系信箱&#xff1a;feixiaoxing 163.com】 soc一般是指跑linux的芯片&#xff0c;而mcu默认是跑rtos的芯片&#xff0c;两者在基本原理方面其实差异不大。只不过&#xff0c;前者由于性能的原…
阅读更多...
离线导出和安装Python库

离线导出和安装Python库

详细介绍&#xff1a;离线导出和安装Python库 常用命令&#xff1a; 生成requirement.txt文件 pip freeze > requirement.txt离线批量下载库 pip download -d packages -r requirement.txt离线批量安装库 pip install --no-index --find-links./ -r requirement.txt
阅读更多...
基于Vue Node.js的电影售票网站的设计与实现(源码+lw+部署文档+讲解)，源码可白嫖!

基于Vue Node.js的电影售票网站的设计与实现(源码+lw+部署文档+讲解)，源码可白嫖!

摘要 互联网技术的成熟和普及&#xff0c;势必会给人们的生活方式带来不同程度的改变。越来越多的经营模式中都少不了线上运营&#xff0c;互联网正强力推动着社会和经济发展。国人对民族文化的自信和不同文化的包容&#xff0c;再加上电影行业的发展&#xff0c;如此繁荣吸引…
阅读更多...
利用 Python 和 AI 技术创作独特的图像艺术作品

利用 Python 和 AI 技术创作独特的图像艺术作品

1. 项目目标 生成艺术作品&#xff1a;利用 AI 模型&#xff08;如 Stable Diffusion&#xff09;生成具有艺术风格的图像。自定义风格&#xff1a;通过文本提示&#xff08;prompt&#xff09;控制图像的艺术风格&#xff08;如赛博朋克、印象派、超现实主义等&#xff09;。…
阅读更多...
XR技术赋能艺术展演｜我的宇宙推动东方美学体验化

XR技术赋能艺术展演｜我的宇宙推动东方美学体验化

本次广州展览现场引入我的宇宙XR体验模块&#xff0c;通过空间计算与动作捕捉技术&#xff0c;让观众在潮玩艺术氛围中体验虚拟互动&#xff0c;打造“看得懂也玩得动”的展演新场景。 作为科技与文化融合的推动者&#xff0c;我的宇宙正在以“体验科技”为媒介&#xff0c;为潮…
阅读更多...
接口的集成测试步骤

接口的集成测试步骤

一、集成测试是什么 ‌接口的集成测试‌是指在软件开发过程中&#xff0c;将各个模块或组件按照设计要求组合在一起&#xff0c;并测试它们之间的接口是否能够正确交互和协同工作的过程。集成测试是软件开发中的一个重要阶段&#xff0c;通常在单元测试之后进行&#xff0c;目的…
阅读更多...
上门服务 APP 30 亿营收商业模式在乌干达的技术赋能与实践

上门服务 APP 30 亿营收商业模式在乌干达的技术赋能与实践

不久前&#xff0c;非洲乌干达出现黑人女技师提供上门足疗服务的消息引发关注。据了解&#xff0c;当地一次40分钟的上门按摩服务仅需约40元人民币&#xff0c;价格仅为国内同类服务的十分之一。这一现象折射出全球健康服务行业正在经历的数字化转型浪潮。 国内领先的上门服务平…
阅读更多...
go语言学习笔记：gin + gorm + mysql 用户增删改查案例入门

go语言学习笔记：gin + gorm + mysql 用户增删改查案例入门

大家好&#xff0c;我是此林。 Golang 语言现在已经成为了编程的趋势&#xff0c;毕竟是大厂背书嘛&#xff0c;Google 研发的。 目前很多云原生项目都是基于 go 来编写的&#xff0c;比如&#xff1a; Kubernetes (K8s)​ 容器编排系统&#xff0c; Docker​ 容器化技术&…
阅读更多...
探索亮数据Web Unlocker API：让谷歌学术网页科研数据 “触手可及”

探索亮数据Web Unlocker API：让谷歌学术网页科研数据 “触手可及”

本文目录 一、引言二、Web Unlocker API 功能亮点三、Web Unlocker API 实战1.配置网页解锁器2.定位相关数据3.编写代码 四、Web Scraper API技术亮点 五、SERP API技术亮点 六、总结 一、引言 网页数据宛如一座蕴藏着无限价值的宝库&#xff0c;无论是企业洞察市场动态、制定…
阅读更多...
SpringBoot整合POI实现Excel文件的导出与导入

SpringBoot整合POI实现Excel文件的导出与导入

使用 Apache POI 操作 Excel文件，系列文章： 《SpringBoot整合POI实现Excel文件的导出与导入》 《SpringMVC实现文件的上传与下载》 《C#使用NPOI导出Excel文件》 《NPOI使用手册》 1、Apache POI 的介绍 Apache POI 是一个基于 Java 的开源库，专为读写 Microsoft Office 格…
阅读更多...
构建用户友好的记账体验 - LedgerX交互设计与性能优化实践

构建用户友好的记账体验 - LedgerX交互设计与性能优化实践

构建用户友好的记账体验 - LedgerX交互设计与性能优化实践 发布日期: 2025-04-16 引言 在财务管理应用领域&#xff0c;技术实力固然重要&#xff0c;但最终决定用户留存的往往是日常使用体验。本文作为LedgerX技术博客的第二篇&#xff0c;将深入探讨我们如何通过精心的交互…
阅读更多...
如何用AI辅助软件产品原型设计及工具推荐

如何用AI辅助软件产品原型设计及工具推荐

以下是针对软件产品原型设计的 AI辅助工具推荐&#xff0c;涵盖国内外主流工具&#xff0c;结合功能特点、优劣势及适用场景分析&#xff0c;并标注是否为国内软件及付费情况&#xff1a; 一、国内工具推荐 1. 墨刀AI&#xff08;MockingBot AI&#xff09; 特点&#xff1a;…
阅读更多...
MySQL的MVCC机制详解

MySQL的MVCC机制详解

1. 什么是MVCC&#xff1f; MVCC&#xff08;Multi-Version Concurrency Control&#xff0c;多版本并发控制&#xff09;是数据库系统中用于实现并发控制的一种技术。它通过保存数据在某个时间点的快照来实现&#xff0c;使得在同一个数据行上可以同时存在多个版本&#xff0…
阅读更多...
最新文章

Copyright @ 2022~2023