前言
在之前使用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应用的接口嘞?待后续慢慢思考;