前言

? 在之前使用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應用的接口嘞？待后續慢慢思考；