之前介绍了 什么是Mcp，以及如何让ClaudeCode来衔接 Mcp，接下来我们来实Mcp 服务的一个核心重头戏——对接数据
一般企业最核心的内容就是数据库，尤其是一些历史悠久的企业，总会存在着大量纷繁复杂的数据库，这些数据分布在不同的系统里，有些数据已经变成了死数据，有些数据业务逻辑已经失效，而有些 有比较活跃~
在以前，如果想要重构项目或者要新增表、导入数据的时候，就必须理清这些数据库内部的逻辑，搞清楚字段之间的关联性，可以想象当你在动辄上万张表，单表动辄上亿数据的 场景下，这项工作将是由多复杂！
如今AI 时代，我们利用AI 来管理我们的数据库自然是最佳选择！AI 的算力足够强大，简单比喻来说，就是在同一时刻有成千上万 “我"的分身去帮我们去干活，一个分身看一张表这样就快多了，更牛的是，这些分身绝不仅仅是只负责自己的事情，它们最种会被中枢神经控制着，最后会汇总到一起，给出我们最想要的答案！
要利用AI 管理我们的私有数据库，那么MCP 就发挥了它 巨大的作用，废话少说，上干货！
一、数据结构
我们以一个简单的demo入手——老师表 —— 课程表 —— 老师课表
-- 老师表CREATE TABLE teacher ( id INT PRIMARY KEY AUTO_INCREMENT, name VARCHAR(50) NOT NULL);-- 课表（课程基础信息）CREATE TABLE course ( id INT PRIMARY KEY AUTO_INCREMENT, name VARCHAR(100) NOT NULL, weekday VARCHAR(10) NOT NULL, -- 周一、周二…… start_time TIME NOT NULL, end_time TIME NOT NULL);-- 关联表：老师-课程CREATE TABLE teacher_course ( teacher_id INT, course_id INT, PRIMARY KEY (teacher_id, course_id), FOREIGN KEY (teacher_id) REFERENCES teacher(id), FOREIGN KEY (course_id) REFERENCES course(id));-- 插入示例数据INSERT INTO teacher (name) VALUES ('张三'), ('李四');INSERT INTO course (name, weekday, start_time, end_time) VALUES('高等数学', '周一', '08:00', '09:40'),('大学物理', '周三', '10:00', '11:40'),('英语阅读', '周三', '14:00', '15:40');INSERT INTO teacher_course VALUES(1, 1), -- 张三教高等数学(1, 2), -- 张三教大学物理(2, 3); -- 李四教英语阅读
我们的目标（tool）有两个：
1、get_teacher_schedule 固定查询 某个老师的课表
输入：老师姓名 teacher_name 输出：该老师的完整课表（课程名、星期、时间）
2、query_custom 自由查询
输入：自然语言描述 natural_language_query 处理：调用 DeepSeek API 将自然语言转为 SQL → 执行 SQL → 返回结果
每个语言实现一个独立 MCP 服务器，通过标准输入输出与 MCP Client（如 Claude Desktop、Continue 等）通信。
三、Python 实现（使用 FastMCP包）
安装依赖
pip install fastmcp pymysql openai pydantic
代码
# mcp_fastmcp_demo.pyimport asyncioimport pymysqlfrom openai import OpenAIfrom mcp.server.fastmcp import FastMCPfrom pydantic import BaseModel, Field # 推荐配合 Pydantic 做输入校验# ---------- 配置 ----------DB_CONFIG = { "host": "localhost", "user": "root", "password": "123456", "database": "tech",}deepseek_client = OpenAI( api_key="xxxxx", base_url="https://api.deepseek.com")# ---------- 创建 MCP 服务器 ----------# stateless_http=True 表示每个请求独立，适合容器化部署mcp = FastMCP("MySQL-MCP-Server", host="0.0.0.0", port=8080, stateless_http=True)# ---------- Tool 1：查询老师课表 ----------# 推荐使用 Pydantic 定义输入，自动获得类型校验和 Schema 生成class TeacherScheduleInput(BaseModel): teacher_name: str = Field(..., description="老师姓名，例如：张三")# 工具一：get_teacher_schedule 查询教师课表@mcp.tool()async def get_teacher_schedule(input: TeacherScheduleInput) -> str: """查询某位老师的完整课表，返回课程名称、星期、上课时间""" conn = pymysql.connect(**DB_CONFIG) try: with conn.cursor() as cursor: sql = """ SELECT c.name, c.weekday, c.start_time, c.end_time FROM teacher t JOIN teacher_course tc ON t.id = tc.teacher_id JOIN course c ON tc.course_id = c.id WHERE t.name = %s ORDER BY FIELD(c.weekday, '周一','周二','周三','周四','周五','周六','周日'), c.start_time """ cursor.execute(sql, (input.teacher_name,)) rows = cursor.fetchall() if not rows: return f"未找到老师 {input.teacher_name} 的课表" result = "\n".join([f"{row[0]} | {row[1]} | {row[2]} - {row[3]}" for row in rows]) return f"【{input.teacher_name} 老师的课表】\n{result}" finally: conn.close()# ---------- Tool 2：自然语言查询 ----------class CustomQueryInput(BaseModel): natural_language_query: str = Field(..., description="自然语言描述的查询需求，例如：查询张三老师周三的课程")# 工具二：query_custom 自由查询@mcp.tool()async def query_custom(input: CustomQueryInput) -> str: """根据自然语言描述自动生成 SQL 并执行查询""" prompt = f"""你是一个MySQL专家。根据以下表结构，将自然语言查询转为SQL语句。 只输出SQL，不要有任何解释或额外文字。 表结构： teacher(id, name) course(id, name, weekday, start_time, end_time) teacher_course(teacher_id, course_id) 自然语言：{input.natural_language_query} SQL:""" response = deepseek_client.chat.completions.create( model="deepseek-chat", messages=[{"role": "user", "content": prompt}], temperature=0 ) sql = response.choices[0].message.content.strip() # TODO:对sql进行校验 conn = pymysql.connect(**DB_CONFIG) try: with conn.cursor() as cursor: cursor.execute(sql) rows = cursor.fetchall() return f"生成SQL: {sql}\n查询结果: {rows}" except Exception as e: return f"SQL执行失败: {e}\n生成SQL: {sql}" finally: conn.close()# ---------- 启动服务器 ----------if __name__ == "__main__": # streamable-http 模式：监听 HTTP 端口，支持远程访问 mcp.run(transport="streamable-http")
FastMCP 是当前 python 提供最优秀的一个 MCP工具包，通过它，我们可以轻松实现任意tool，想要面试的童鞋，一定要好好练习使用这个包
第一个tool很传统，我们只需要 写好固定的查询逻辑，然后让AI 触发关键词调用即可，而第二个案例，我们需要借助AI 来 通过用户提示词让AI来组织成对应的SQL，然后系统再调用SQL，比较灵活，这里值得注意的是，一定要做出限制，比如不允许出现update、delete或create 等关键词，禁止对数据库做出修改等指令
六、运行测试
通用配置 确保 MySQL 服务已启动，并执行上面的建表 + 示例数据。
替换代码中的数据库密码、DeepSeek API Key。
使用支持 MCP 的客户端（如 Claude Desktop 或 mcp-cli 或阅读上篇文章，配置到 Claude Code中）加载对应服务器。
测试用例
查询老师课表 输入：调用 get_teacher_schedule，teacher_name = “张三” 返回：张三老师的课表（高等数学周一、大学物理周三）
自然语言查询 输入：“查询张三老师周三的课程” DeepSeek 自动生成 SQL 并执行，返回大学物理的信息。
七、总结
这里只给出了很简单的 两个tool，实际业务肯定不会这么简单，核心重点还是这个tool服务上，后续还会继续退出更多相关内容，如Java /go /nodejs等实现MCP 管理数据库，以及更多复杂的业务场景！
当然 McpServer 不仅仅能用在 claude code上，我还会分享 McpServer 更多的用处以及有价值的第三方McpServer!
关注我，分享更多干货！
10个月宝宝每天需要喝多少奶粉？
