网中看 网中看 28天前

使用 FastMCP 编写一个 MySQL MCP Server

python

在日常工作中，当我们针对某个业务场景不知 SQL 如何编写时，或在应用程序中找到一些慢 SQL 需要优化而不知所措时，通常会询问 AI 助手。但我们若不提供任何上下文，仅仅是用一句话将业务场景描述给 AI 助手让其实现，或贴一段很长的 SQL 让 AI 助手来优化，其给出的指导意见的质量通常会大打折扣。

所以，要让 AI 助手给出高效的指导意见，需要提供充分的上下文。在数据库场景下，最重要的上下文就是表结构。

而诸如表结构的上下文如何提供给 AI 助手呢？手动把数据库中的表结构抓取出来放到文件里？然后提问时，附上这些文件？这个方法不是不行，但效率实在是太低。

由前文「MCP 是什么？它是如何工作的？」可以知道，MCP 是大语言模型连接外部工具或服务的桥梁，MCP Server 就是用来给大语言模型提供上下文的。

所以，假设我们使用的数据库为 MySQL，那就可以编写一个 MySQL MCP Server 来给使用大语言模型的 AI 助手提供上下文。这样，拥有了足够上下文的 AI 助手在回答编写 SQL 或优化 SQL 相关的问题时就会更加准确可靠。

本文即是针对这个设想的实现：首先会使用 Python 语言编写一个 MySQL MCP Server，然后在 AI 助手 Windsurf 中进行使用。
1 编写 MySQL MCP Server

本部分将使用 Python 的 FastMCP 框架来编写一个 MySQL MCP Server。

FastMCP 是 Python 中比较流行的 MCP 应用构建框架，其简单易用、可用于生产环境、内置鉴权部署等工具。

下面即是编写好的 MySQL MCP Server 文件 mysql_mcp_server.py 的代码：

  1. #! /usr/bin/env python3
  2. import os
  3. import traceback
  4. from typing import Any, Dict, List
  6. import pymysql
  7. from fastmcp import FastMCP
  8. from pymysql import Connection
  10. mcp = FastMCP("mysql-mcp-server")
  13. def get_db_config() -> Dict[str, Any]:
  14.     """Read database config from environment variables and return a dictionary."""
  15.     return {
  16.         "host": os.getenv("MYSQL_HOST", "localhost"),
  17.         "port": int(os.getenv("MYSQL_PORT", "3306")),
  18.         "user": os.getenv("MYSQL_USER", "root"),
  19.         "password": os.getenv("MYSQL_PASSWORD", ""),
  20.         "database": os.getenv("MYSQL_DATABASE", ""),
  21.         "cursorclass": pymysql.cursors.DictCursor
  22.     }
  25. def get_connection() -> Connection[Any]:
  26.     """Get a database connection"""
  27.     return pymysql.connect(**get_db_config())
  30. @mcp.tool("list_tables", description="List all the tables in the database")
  31. def list_tables() -> List[str]:
  32.     conn = get_connection()
  33.     try:
  34.         with conn.cursor() as cur:
  35.             cur.execute("SHOW TABLES")
  36.             tables = [row[next(iter(row))] for row in cur.fetchall()]
  37.             return tables
  38.     except Exception as e:
  39.         print("Error in list tables:", e)
  40.         traceback.print_exc()
  41.         raise RuntimeError(f"list tables failed: {e}")
  42.     finally:
  43.         conn.close()
  46. @mcp.tool("describe_table", description="Query the table structure for a specified table")
  47. def describe_table(table_name: str) -> List[Dict[str, Any]]:
  48.     conn = get_connection()
  49.     try:
  50.         with conn.cursor() as cur:
  51.             cur.execute(f"DESCRIBE {table_name}")
  52.             columns = cur.fetchall()
  53.             return list(columns) if columns else []
  54.     except Exception as e:
  55.         print("Error in describe table:", e)
  56.         traceback.print_exc()
  57.         raise RuntimeError(f"describe table failed: {e}")
  58.     finally:
  59.         conn.close()
  62. @mcp.tool("explain_sql", description="Explain a SQL query")
  63. def explain_sql(query: str) -> List[Dict[str, Any]]:
  64.     conn = get_connection()
  65.     try:
  66.         with conn.cursor() as cur:
  67.             cur.execute(f"EXPLAIN {query}")
  68.             result = cur.fetchall()
  69.             return result
  70.     except Exception as e:
  71.         print("Error in explain sql:", e)
  72.         traceback.print_exc()
  73.         raise RuntimeError(f"explain sql failed: {e}")
  74.     finally:
  75.         conn.close()
  78. @mcp.tool("run_sql", description="Run SELECT SQL query")
  79. def run_sql(query: str, limit: int = 1000) -> List[Dict[str, Any]]:
  80.     if not query.strip().lower().startswith("select"):
  81.         raise RuntimeError("Only SELECT queries are supported")
  83.     conn = get_connection()
  84.     try:
  85.         with conn.cursor() as cur:
  86.             cur.execute(f"{query} LIMIT {limit}")
  87.             result = cur.fetchall()
  88.             return result
  89.     except Exception as e:
  90.         print("Error in run sql:", e)
  91.         traceback.print_exc()
  92.         raise RuntimeError(f"run sql failed: {e}")
  93.     finally:
  94.         conn.close()
  97. if __name__ == "__main__":
  98.     mcp.run()
  99.     # mcp.run(transport="http", host="0.0.0.0", port=8000)

可以看到，使用 FastMCP 编写 MCP Server 非常的简单。

在上述代码中：

  1. 首先使用 mcp = FastMCP("mysql-mcp-server") 的方式创建了一个 mcp 实例；
  3. 然后创建了两个普通函数 get_db_config() 和 get_connection()，分别用于从环境变量中获取数据库连接信息和获取数据库连接；
  5. 接着创建了四个函数，并在函数上方加上了装饰器 @mcp.tool()，表示它们是四个 MCP 工具：list_tables、describe_table、explain_sql 和 run_sql，分别用于列出所有的表、查询给定表的表结构、查询 SQL 的执行计划和执行查询语句；
  7. 最后使用 if __name__ == "__main__" 判断程序作为主程序运行时调用 mcp.run()，这样执行 mysql_mcp_server.py 时一个 Stdio（标准输入输出）方式的 MCP Server 就启动了。

2 在 AI 助手 Windsurf 中进行使用

下面，我们尝试在 AI 助手 Windsurf 中使用一下上面编写好的 MCP Server。
2.1 使用本地启动的 MCP Server

在 Windsurf 中使用 MCP Server 与在 VS Code、Cursor 中类似，首先需要编写一个 mcp json 配置文件，Windsurf 的 mcp_config.json 配置文件需要放在 ~/.codeium/windsurf/ 目录下。

下面就是我们在 mcp_config.json 配置文件填入的内容：

{
“mcpServers”: {
“mysql-mcp-server”: {
“command”: “python3”,
“args”: [
“/Users/larry/python-exercises/mysql-mcp-server-demo/mysql_mcp_server.py”
],
“env”: {
“MYSQL_HOST”: “localhost”,
“MYSQL_PORT”: “3306”,
“MYSQL_USER”: “root”,
“MYSQL_PASSWORD”: “root”,
“MYSQL_DATABASE”: “test”
}
}
}
}

我们指定了 MCP Server 的名字为 mysql-mcp-server；启动命令为 python3；启动参数为上述源码文件 mysql_mcp_server.py 的位置；环境变量值为 MYSQL 的具体连接信息。

MCP Server 配置好后，即可以在 Windsurf 编辑器的对话框中使用了。

我们尝试发送如下列出所有数据库表的指令：

2.2 使用远程启动的 MCP Server

上面的 MCP Server 需要在本地启动，传输模式为 Stdio；如果我们想将 MCP Server 部署在服务器，该如何适配呢？

MCP 还有一种传输模式，叫 Streamable HTTP（流式 HTTP），使用该模式就可以将 MCP Server 部署在服务器，然后在 AI 助手配置一个 URL 就可以使用了。

我们尝试在本地将 mysql_mcp_server.py 的入口代码改写为如下方式并启动：

if name == “main“:
mcp.run(transport=”http”, host=”0.0.0.0”, port=8000)

然后将 Windsurf 的 mcp_config.json 配置改用如下方式，就能以流式 HTTP 的方式和 MCP Server 交互了，功能与上述 Stdio 无异。

{
“mcpServers”: {
“mysql-mcp-server”: {
“url”: “http://localhost:8000/mcp“
}
}
}

3 小结

综上，本文首先针对 SQL 编写和 SQL 优化的场景，提出不提供上下文的情况下 AI 助手给出的参考结果准确性和质量受限；然后提出编写一个 MCP Server 给大语言模型补充上下文的想法；接着使用 Python 语言编写一个 MySQL MCP Server 并在 AI 助手 Windsurf 中进行了配置和使用。

源文出处：https://leileiluoluo.com/posts/writing-a-mysql-mcp-server.html