日常开发中，你是不是也经常遇到这种情况：想让 AI 帮你写一段 SQL，或者优化一条慢查询，但光靠嘴描述业务逻辑，AI 给出来的结果总是“差点意思”？问题出在哪？缺上下文。
尤其是数据库相关的任务，最关键的上下文就是表结构。如果你不告诉 AI 表里有哪些字段、主键外键怎么设的，它只能凭空猜——那结果自然没法直接用。
手动复制建表语句粘贴过去？太麻烦了。有没有更聪明的办法？有！这就是 MCP（Model Context Protocol） 要解决的问题。
程序员导航
优网导航旗下整合全网优质开发资源，一站式IT编程学习与工具大全网站
什么是 MCP？简单说就是给大模型“接外挂”
MCP 是一种协议，能让大语言模型（LLM）安全地调用外部工具或服务。你可以把它理解成 LLM 的“插件系统”。通过 MCP Server，AI 助手就能动态获取数据库结构、执行查询、分析执行计划等，而不需要你手动喂数据。
本文就手把手教你用 Python + FastMCP 框架，写一个 MySQL MCP Server，让它在 AI 助手（比如 Windsurf）里跑起来，真正实现“懂你库”的智能 SQL 辅助。
为什么选 FastMCP？
FastMCP 是 Python 社区里一个轻量又实用的 MCP 实现框架。它：
- 写起来简单，几行装饰器就能暴露工具；
- 支持生产级部署；
- 内置鉴权、日志、传输模式切换等功能。
对我们这种想快速验证想法的开发者来说，再合适不过。
AI 工具导航
优网导航旗下AI工具导航，精选全球千款优质 AI 工具集
核心代码：四个工具搞定数据库交互
下面这个 mysql_mcp_server.py 文件，就是我们今天的主角。它通过四个工具函数，把 MySQL 的关键能力暴露给 AI：
#! /usr/bin/env python3
import os
import traceback
from typing import Any, Dict, List
import pymysql
from fastmcp import FastMCP
from pymysql import Connection
# 创建 FastMCP 实例，名字叫 "mysql-mcp-server"
mcp = FastMCP("mysql-mcp-server")
def get_db_config() -> Dict[str, Any]:
"""从环境变量读取数据库连接配置"""
return {
"host": os.getenv("MYSQL_HOST", "localhost"),
"port": int(os.getenv("MYSQL_PORT", "3306")),
"user": os.getenv("MYSQL_USER", "root"),
"password": os.getenv("MYSQL_PASSWORD", ""),
"database": os.getenv("MYSQL_DATABASE", ""),
"cursorclass": pymysql.cursors.DictCursor # 返回字典格式结果，方便处理
}
def get_connection() -> Connection[Any]:
"""根据配置建立 MySQL 连接"""
return pymysql.connect(**get_db_config())
💡 小贴士：这里所有数据库参数都从环境变量读取，既安全又灵活，适合本地调试和部署。
接下来是四个核心工具，每个都用 @mcp.tool 装饰器注册，AI 就能自动发现并调用它们：
1. 列出所有表（list_tables）
@mcp.tool("list_tables", description="List all the tables in the database")
def list_tables() -> List[str]:
conn = get_connection()
try:
with conn.cursor() as cur:
cur.execute("SHOW TABLES")
# SHOW TABLES 返回的列名是动态的（如 Tables_in_test），所以取第一个 key 的值
tables = [row[next(iter(row))] for row in cur.fetchall()]
return tables
except Exception as e:
print("Error in list tables:", e)
traceback.print_exc()
raise RuntimeError(f"list tables failed: {e}")
finally:
conn.close()
2. 查看表结构（describe_table）
@mcp.tool("describe_table", description="Query the table structure for a specified table")
def describe_table(table_name: str) -> List[Dict[str, Any]]:
conn = get_connection()
try:
with conn.cursor() as cur:
# 注意：这里没做 SQL 注入防护，仅限可信环境使用
cur.execute(f"DESCRIBE {table_name}")
columns = cur.fetchall()
return list(columns) if columns else []
except Exception as e:
print("Error in describe table:", e)
traceback.print_exc()
raise RuntimeError(f"describe table failed: {e}")
finally:
conn.close()
⚠️ 注意：
f"DESCRIBE {table_name}"这种拼接方式在生产环境有风险。如果要上线，建议加白名单校验或改用参数化方式（虽然 MySQL 的 DESCRIBE 不支持参数化）。
3. 分析 SQL 执行计划（explain_sql）
@mcp.tool("explain_sql", description="Explain a SQL query")
def explain_sql(query: str) -> List[Dict[str, Any]]:
conn = get_connection()
try:
with conn.cursor() as cur:
cur.execute(f"EXPLAIN {query}")
result = cur.fetchall()
return result
except Exception as e:
print("Error in explain sql:", e)
traceback.print_exc()
raise RuntimeError(f"explain sql failed: {e}")
finally:
conn.close()
这个对优化慢 SQL 特别有用。AI 可以先让你提供原始 SQL，然后主动调用 explain_sql 看执行计划，再给出索引建议或重写方案。
4. 安全执行 SELECT 查询（run_sql）
@mcp.tool("run_sql", description="Run SELECT SQL query")
def run_sql(query: str, limit: int = 1000) -> List[Dict[str, Any]]:
# 限制只允许 SELECT，防止误删改数据
if not query.strip().lower().startswith("select"):
raise RuntimeError("Only SELECT queries are supported")
conn = get_connection()
try:
with conn.cursor() as cur:
cur.execute(f"{query} LIMIT {limit}") # 自动加 LIMIT 防止查爆
result = cur.fetchall()
return result
except Exception as e:
print("Error in run sql:", e)
traceback.print_exc()
raise RuntimeError(f"run sql failed: {e}")
finally:
conn.close()
🔒 安全设计：只允许
SELECT，且自动加LIMIT 1000，避免拖垮数据库。免费在线工具导航
优网导航旗下整合全网优质免费、免注册的在线工具导航大全
最后，启动服务：
if __name__ == "__main__":
mcp.run() # 默认使用 Stdio 模式（适合本地集成）
# 如果要部署成 HTTP 服务，可改为：
# mcp.run(transport="http", host="0.0.0.0", port=8000)
在 Windsurf 里配置并使用
Windsurf（或其他支持 MCP 的 AI 编辑器）需要一个配置文件来知道怎么调你的 Server。
在 ~/.codeium/windsurf/mcp_config.json 里填入：
{
"mcpServers": {
"mysql-mcp-server": {
"command": "python3",
"args": ["/path/to/your/mysql_mcp_server.py"],
"env": {
"MYSQL_HOST": "localhost",
"MYSQL_PORT": "3306",
"MYSQL_USER": "root",
"MYSQL_PASSWORD": "root",
"MYSQL_DATABASE": "test"
}
}
}
}
配置完重启 Windsurf，就可以直接对话了。
实测效果
- 问：“用 mysql-mcp-server 查询我的数据库里有哪些表？”
→ AI 自动调list_tables，返回表名列表。 
问：“查一下 moment 表的结构。”
→ AI 调describe_table("moment")，返回字段详情。- 问：“下面这段 SQL 很慢，帮我优化……”
→ AI 先分析逻辑，再主动问：“要不要我跑个 EXPLAIN 看看执行计划？”
→ 你回“要”，它就调explain_sql，结合结果给出具体优化建议。
问：“查一下 moment 表的结构。”
实测下来，有了真实表结构和执行计划，AI 的建议靠谱多了，不再是泛泛而谈。
想部署到服务器？改两行就行
上面默认是 Stdio 模式，适合本地 IDE 集成。如果你想把 MCP Server 部署成独立服务，只需：
- 修改启动代码：
if __name__ == "__main__": mcp.run(transport="http", host="0.0.0.0", port=8000) - 更新
mcp_config.json，改成 URL 方式：{ "mcpServers": { "mysql-mcp-server": { "url": "http://your-server-ip:8000/mcp" } } }
这样，任何地方的 AI 助手只要能访问这个 URL，就能用你的 MySQL 上下文了。
总结
通过 FastMCP 写一个 MySQL MCP Server，成本极低，但收益明显：
- 让 AI 写 SQL 时“看得见”你的表结构；
- 优化慢查询时能结合真实执行计划；
- 所有操作可控、安全、可审计。
这比你每次手动粘贴 DDL 或 EXPLAIN 结果高效太多。尤其适合团队共享数据库上下文，或者集成到内部开发工具链中。
相关文章
