【大模型】基于MCP的mysql 服务构建及使用（python语言）

前言

​ 在之前使用dify来编排AI智能体，有这样的一个场景，希望智能体能自动读取数据库数据，获得统计数据（问数），最终生成报告。

​ 当时实现思路是，通过知识库告诉大模型相关表的字段定义，然后让大模型根据提示词生成对应统计指标SQL语句，然后调用mysql工具去执行，返回结果。这其中，需要整理一堆表定义的知识库，且最终效果也不好。

​ MCP（Model Context Protocol，模型上下文协议 ）协议出现后，大量MCP服务涌出，其中mysql 类的MCP服务，使用cline可以得出不错的问数效果。但基本都是Typescript实现的，且安全性无法保证。在实际使用中，更愿意自己去维护一个本地MCP服务。该文档即是探究使用python开发调试本地mysql MCP服务的方法。

功能分析

• 执行查询SQL功能

  由大模型生成相关问题查询SQL语句，发送给MCP服务执行，并返回结果给大模型

• 查询指定表的模式

  要想让大模型生成正确的SQL，需要告诉大模型指定表的结构模式

• 查询指定数据库所有表

  想要获取指定表名，需要先列出所有的表名

初始化项目环境

uv init mysql-mcp
cd mysql-mcp
# 创建mcp服务虚拟环境
uv venv
# 激活该环境
.venv\Scripts\activate

代码实现

安装依赖包

pip install mysql-connector-python

uv add "mcp[cli]"

代码如下：

mysql-mcp.py

from mcp.server.fastmcp import FastMCP
from config import DB_CONFIG
import pymysql
import utils# 初始化mcp服务
mcp = FastMCP("mysql-mcp")@mcp.tool()
def list_tables():"""获取数据库所有表名"""try:mydb = pymysql.connect(**DB_CONFIG)mycursor = mydb.cursor()mycursor.execute("SHOW TABLES")tables = mycursor.fetchall()table_names = [table[0] for table in tables]return table_namesexcept pymysql.Error as err:print(f"MySQL Error: {err}")return {"error": str(err)}finally:if 'mycursor' in locals():mycursor.close()if 'mydb' in locals():mydb.close()@mcp.tool()
def describ_table(table_name:str):"""返回表的结构"""try:mydb = pymysql.connect(**DB_CONFIG)mycursor = mydb.cursor()mycursor.execute("SELECT column_name, data_type FROM information_schema.columns WHERE table_name = %s",table_name)table_desc = mycursor.fetchall()return table_descexcept pymysql.Error as err:print(f"MySQL Error: {err}")return {"error": str(err)}finally:if 'mycursor' in locals():mycursor.close()if 'mydb' in locals():mydb.close()@mcp.tool()
def execute_sql(sql:str):"""执行SQL语句,返回查询结果"""try:mydb = pymysql.connect(**DB_CONFIG)mycursor = mydb.cursor()mycursor.execute(sql)# 获取列名column_names = [desc[0] for desc in mycursor.description]if mycursor.rowcount == 0:mydb.commit()return {"message": "执行成功，无返回结果"}else:return utils.data_transformer(column_names,mycursor.fetchall())except pymysql.Error as err:print(f"MySQL Error: {err}")return {"error": str(err)}finally:if 'mycursor' in locals():mycursor.close()if 'mydb' in locals():mydb.close()if __name__ == "__main__":print("启动mysql-mcp服务")mcp.run(transport="stdio")

config.py,配置数据库连接

DB_CONFIG = {"host": "127.0.0.1","port": 3306,"user": "root","password": "XXXX@123456","database": "XXX",}

调试

调试bug

在写完代码后，可以使用如下命令去调试是否存在语法错误

python ./mysql-mcp.py

可以看到mcp包未安装

mysql 包未安装

目录名和mysql包名重复了，修改目录名

经过多次安装，还是提示不存在。在这个目录层次下，使用conda 虚拟了一个python环境，python版本为3.10，在mysql目录下又使用uv venv虚拟了一个环境，但该虚拟环境是用的conda环境下的python包，所以有点乱；

于是直接将uv venv虚拟环境删掉，直接新建目录mysqlmcp,然后复制如上的代码。

执行

mcp dev ./mysql-mcp.py

这时将会启动成功。浏览器输入地址 http://127.0.0.1:6274

输入如下参数：

command

uv

Arguments

run --with mcp mcp run ./mysql-mcp.py

Proxy Session Token 从启动的控制台Session Token 复制

点击连接，即可连接成功。

调试MCP工具

连上去后就可以开始调试Tools了。

先点击list Tools列出MCP服务提供的工具，如上图。

点击list_tables,执行run tools

可以看出该工具成功返回结果，数据库中的表，history面板展示了调用工具的请求和返回通信信息。

这里的数据库连接信息是配置在代码侧的。

其他的工具调试类似。但存在一个问题，这种调试方式是没法debug的，没法单步调试的，所以推荐可以对工具函数，单独写python代码进行调试。调试好再同步到MCP服务里。

使用cline连接

开发好MCP服务，就可以用cline去用了。

在 mcp inspector 界面左侧面板复制 servers file，然后粘贴到cline的mcp服务配置文档里(需要调整下)，调整后的如下：

"mysql-server": {"disabled": false,"timeout": 60,"type": "stdio","command": "uv","args": ["run","--with","mcp","mcp","run","D:\\source\\mcp\\mysqlmcp\\mysql-mcp.py"]}

点击连接报错

后面证实这其实不是报错，只是连接MCP服务的日志信息。

现在是已经正常连接了。

执行数据统计问答

正常连接后，向其提问：

查看数据库有多少个党员，并按年龄段统计展示

可以看到cline利用LLM，已经为这个问题做好了规划，一共5步。后续就是按照这5步去执行。

首先，cline识别出了mysql-server的list_tables工具，获取数据库所有表

其次，调用describ_table工具去获取表的字段描述

再次，对字段进行分析，写SQL语句

执行SQL报错，自动修改SQL，

最终执行成功，返回结果。

经验总结

1、MCP服务整体开发思路大致是 使用uv初始化环境，写MCP服务代码，代码单点调试，启用mcp inspector调试、使用cline连接，具体使用；uv虚拟环境有时候感觉需要，有时候又感觉不需要，后续整个全新的项目去研究一下；

2、应用过程中，我在想，使用cline去调用mcp服务的工具，和基于http协议调用接口有啥区别嘞，为啥不能直接去调用http应用的接口嘞？待后续慢慢思考；