一、概述
使用python开发，最好的框架是fastmcp，github连接：https://github.com/jlowin/fastmcp
2025 年 5 月 9 日，fastmcp发布v2.3.0版本，正式支持Streamable HTTP
终于等到官方支持了！
注意：2.3.0版本有bug，目前最新版本已经修复了
升级到最新版本
pip install --upgrade fastmcp
关于Streamable HTTP的介绍，请参考链接：https://www.cnblogs.com/xiao987334176/p/18845151
这里就不再重复了
二、Streamable HTTP MCP应用
官方demo
server.py
from fastmcp import FastMCP mcp = FastMCP("Demo 🚀") @mcp.tool() def add(a: int, b: int) -> int: """Add two numbers""" return a + b if __name__ == "__main__": mcp.run(transport="streamable-http", host="0.0.0.0", port=8000, path="/mcp")
通过以上8行代码，就简单实现了Streamable HTTP MCP应用。
获取服务器公网ip
通过一个在实际生产环境中，使用的功能，来演示一个Streamable HTTP MCP应用
server.py
import json import requests from fastmcp import FastMCP mcp = FastMCP("Demo 🚀") @mcp.tool() def get_public_ip_address() -> str: """ 获取服务器公网 IP 地址 返回: str: 当前网络的公网 IP 地址 """ try: response = requests.get("http://ip-api.com/json") response.raise_for_status() # 检查 HTTP 请求是否成功 content = json.loads(response.text) return content.get("query", "Unknown IP") # 提供默认值以防字段缺失 except requests.RequestException as e: print(f"请求错误: {e}") return "Request Failed" except json.JSONDecodeError as e: print(f"JSON 解码错误: {e}") return "Invalid Response" if __name__ == "__main__": # mcp.run() mcp.run(transport="streamable-http", host="0.0.0.0", port=9000, path="/mcp")
运行代码
python server.py
输出：
[05/12/25 10:03:55] INFO Starting server "public_ip_address"... server.py:202 INFO: Started server process [43312] INFO: Waiting for application startup. INFO: Application startup complete. INFO: Uvicorn running on http://0.0.0.0:9000 (Press CTRL+C to quit)
Cherry Studio测试
请确保Cherry Studio版本是最新的，因为新版本，增加了Streamable HTTP支持
添加mcp服务器
名称：public_ip_address_mcp
类型：Streamable HTTP
url：http://localhost:9000/mcp
添加成功后，点击工具，可以看到工具方法
创建
添加
选择助手
选择
提问公网ip，效果如下：
验证一下公网ip是否正确，打开网页
结果是正确的，没问题。
三、SSE转换为Streamable HTTP
代码改造
在前面的文章中，写的mysql应用是SSE模式，链接：https://www.cnblogs.com/xiao987334176/p/18827297
把代码拷贝过来，只需要修改2行代码，就可以无缝转换为Streamable HTTP
将以下2行
mcp = FastMCP("operateMysql", host="0.0.0.0", port=9000) mcp.run(transport="sse")
修改为：
mcp = FastMCP("operateMysql") mcp.run(transport="streamable-http", host="0.0.0.0", port=9000, path="/mcp")
mysql_mcp_server_pro应用，完整代码如下：
server.py
from fastmcp import FastMCP from mysql.connector import connect, Error import os mcp = FastMCP("operateMysql") def get_db_config(): """从环境变量获取数据库配置信息 返回: dict: 包含数据库连接所需的配置信息 - host: 数据库主机地址 - port: 数据库端口 - user: 数据库用户名 - password: 数据库密码 - database: 数据库名称 异常: ValueError: 当必需的配置信息缺失时抛出 """ config = { "host": os.getenv("MYSQL_HOST", "localhost"), "port": int(os.getenv("MYSQL_PORT", "3306")), "user": os.getenv("MYSQL_USER"), "password": os.getenv("MYSQL_PASSWORD"), "database": os.getenv("MYSQL_DATABASE"), } print(config) if not all( [ config["host"], config["port"], config["user"], config["password"], config["database"], ] ): raise ValueError("缺少必需的数据库配置") return config @mcp.tool() def execute_sql(query: str) -> list: """执行SQL查询语句 参数: query (str): 要执行的SQL语句，支持多条语句以分号分隔 返回: list: 包含查询结果的TextContent列表 - 对于SELECT查询：返回CSV格式的结果，包含列名和数据 - 对于SHOW TABLES：返回数据库中的所有表名 - 对于其他查询：返回执行状态和影响行数 - 多条语句的结果以"---"分隔 异常: Error: 当数据库连接或查询执行失败时抛出 """ config = get_db_config() try: with connect(**config) as conn: with conn.cursor() as cursor: statements = [stmt.strip() for stmt in query.split(";") if stmt.strip()] results = [] for statement in statements: try: cursor.execute(statement) # 检查语句是否返回了结果集 (SELECT, SHOW, EXPLAIN, etc.) if cursor.description: columns = [desc[0] for desc in cursor.description] rows = cursor.fetchall() # 将每一行的数据转换为字符串，特殊处理None值 formatted_rows = [] for row in rows: formatted_row = [ "NULL" if value is None else str(value) for value in row ] formatted_rows.append(",".join(formatted_row)) # 将列名和数据合并为CSV格式 results.append( "n".join([",".join(columns)] + formatted_rows) ) # 如果语句没有返回结果集 (INSERT, UPDATE, DELETE, etc.) else: conn.commit() # 只有在非查询语句时才提交 results.append(f"查询执行成功。影响行数: {cursor.rowcount}") except Error as stmt_error: # 单条语句执行出错时，记录错误并继续执行 results.append( f"执行语句 '{statement}' 出错: {str(stmt_error)}" ) # 可以在这里选择是否继续执行后续语句，目前是继续 return ["n---n".join(results)] except Error as e: print(f"执行SQL '{query}' 时出错: {e}") return [f"执行查询时出错: {str(e)}"] @mcp.tool() def get_table_name(text: str) -> list: """根据表的中文注释搜索数据库中的表名 参数: text (str): 要搜索的表中文注释关键词 返回: list: 包含查询结果的TextContent列表 - 返回匹配的表名、数据库名和表注释信息 - 结果以CSV格式返回，包含列名和数据 """ config = get_db_config() sql = "SELECT TABLE_SCHEMA, TABLE_NAME, TABLE_COMMENT " sql += f"FROM information_schema.TABLES WHERE TABLE_SCHEMA = '{config['database']}' AND TABLE_COMMENT LIKE '%{text}%';" return execute_sql(sql) @mcp.tool() def get_table_desc(text: str) -> list: """获取指定表的字段结构信息 参数: text (str): 要查询的表名，多个表名以逗号分隔 返回: list: 包含查询结果的列表 - 返回表的字段名、字段注释等信息 - 结果按表名和字段顺序排序 - 结果以CSV格式返回，包含列名和数据 """ config = get_db_config() # 将输入的表名按逗号分割成列表 table_names = [name.strip() for name in text.split(",")] # 构建IN条件 table_condition = "','".join(table_names) sql = "SELECT TABLE_NAME, COLUMN_NAME, COLUMN_COMMENT " sql += ( f"FROM information_schema.COLUMNS WHERE TABLE_SCHEMA = '{config['database']}' " ) sql += f"AND TABLE_NAME IN ('{table_condition}') ORDER BY TABLE_NAME, ORDINAL_POSITION;" return execute_sql(sql) @mcp.tool() def get_lock_tables() -> list: """ 获取当前mysql服务器InnoDB 的行级锁 返回: list: 包含查询结果的TextContent列表 """ sql = """SELECT p2.`HOST` AS 被阻塞方host, p2.`USER` AS 被阻塞方用户, r.trx_id AS 被阻塞方事务id, r.trx_mysql_thread_id AS 被阻塞方线程号, TIMESTAMPDIFF(SECOND, r.trx_wait_started, CURRENT_TIMESTAMP) AS 等待时间, r.trx_query AS 被阻塞的查询, l.OBJECT_NAME AS 阻塞方锁住的表, m.LOCK_MODE AS 被阻塞方的锁模式, m.LOCK_TYPE AS '被阻塞方的锁类型(表锁还是行锁)', m.INDEX_NAME AS 被阻塞方锁住的索引, m.OBJECT_SCHEMA AS 被阻塞方锁对象的数据库名, m.OBJECT_NAME AS 被阻塞方锁对象的表名, m.LOCK_DATA AS 被阻塞方事务锁定记录的主键值, p.`HOST` AS 阻塞方主机, p.`USER` AS 阻塞方用户, b.trx_id AS 阻塞方事务id, b.trx_mysql_thread_id AS 阻塞方线程号, b.trx_query AS 阻塞方查询, l.LOCK_MODE AS 阻塞方的锁模式, l.LOCK_TYPE AS '阻塞方的锁类型(表锁还是行锁)', l.INDEX_NAME AS 阻塞方锁住的索引, l.OBJECT_SCHEMA AS 阻塞方锁对象的数据库名, l.OBJECT_NAME AS 阻塞方锁对象的表名, l.LOCK_DATA AS 阻塞方事务锁定记录的主键值, IF(p.COMMAND = 'Sleep', CONCAT(p.TIME, ' 秒'), 0) AS 阻塞方事务空闲的时间 FROM performance_schema.data_lock_waits w INNER JOIN performance_schema.data_locks l ON w.BLOCKING_ENGINE_LOCK_ID = l.ENGINE_LOCK_ID INNER JOIN performance_schema.data_locks m ON w.REQUESTING_ENGINE_LOCK_ID = m.ENGINE_LOCK_ID INNER JOIN information_schema.INNODB_TRX b ON b.trx_id = w.BLOCKING_ENGINE_TRANSACTION_ID INNER JOIN information_schema.INNODB_TRX r ON r.trx_id = w.REQUESTING_ENGINE_TRANSACTION_ID INNER JOIN information_schema.PROCESSLIST p ON p.ID = b.trx_mysql_thread_id INNER JOIN information_schema.PROCESSLIST p2 ON p2.ID = r.trx_mysql_thread_id ORDER BY 等待时间 DESC;""" return execute_sql(sql) if __name__ == "__main__": mcp.run(transport="streamable-http", host="0.0.0.0", port=9000, path="/mcp")
编译镜像
修改Dockerfile，升级fastmcp
FROM python:3.13.3-alpine3.21 ADD . /app RUN pip3 install --upgrade pip -i https://pypi.tuna.tsinghua.edu.cn/simple && pip3 install mysql-connector-python fastmcp -i https://pypi.tuna.tsinghua.edu.cn/simple && pip3 install --upgrade fastmcp WORKDIR /app EXPOSE 9000 ENTRYPOINT ["python3","/app/server.py"]
编译镜像
docker build -t mysql_mcp_server_pro:v1 .
使用docker-compose启动
docker-compose.yaml
services: mysql_mcp_server_pro: image: mysql_mcp_server_pro:v1 container_name: mysql_mcp_server_pro ports: - "9090:9000" environment: MYSQL_HOST: "192.168.20.128" MYSQL_PORT: "3306" MYSQL_USER: "root" MYSQL_PASSWORD: "abcd@1234" MYSQL_DATABASE: "test" TZ: Asia/Shanghai restart: always
注意修改mysql相关环境变量
运行
docker-compose up -d
Cherry Studio测试
添加MCP服务器
添加智能体mysql8
提示词和以前一样
使用中文回复。 当用户提问中涉及学生、教师、成绩、班级、课程等实体时，需要使用 MySQL MCP 进行数据查询和操作，表结构说明如下： # 学生管理系统数据库表结构说明 ## 1. 教师表 (teachers) | 字段名 | 类型 | 描述 | 约束 | 示例 | |--------|------|------|------|------| | id | varchar | 教师ID | 主键 | "T001" | | name | varchar | 教师姓名 | 必填 | "张建国" | | gender | enum | 性别 | "男"或"女" | "男" | | subject | varchar | 教授科目 | 必填 | "数学" | | title | varchar | 职称 | 必填 | "教授" | | phone | varchar | 联系电话 | 必填 | "13812345678" | | office | varchar | 办公室位置 | 必填 | "博学楼301" | | wechat | varchar | 微信(可选) | 可选 | "lily_teacher" | | isHeadTeacher | enum | 是否为班主任，"true"或"false" | 可选 | true | ## 2. 班级表 (classes) | 字段名 | 类型 | 描述 | 约束 | 示例 | |--------|------|------|------|------| | id | varchar | 班级ID | 主键 | "202301" | | className | varchar | 班级名称 | 必填 | "2023级计算机1班" | | grade | int | 年级 | 必填 | 2023 | | headTeacherId | varchar | 班主任ID | 外键(teachers.id) | "T003" | | classroom | varchar | 教室位置 | 必填 | "1号楼302" | | studentCount | int | 学生人数 | 必填 | 35 | | remark | varchar | 备注信息 | 可选 | "市级优秀班集体" | ## 3. 课程表 (courses) | 字段名 | 类型 | 描述 | 约束 | 示例 | |--------|------|------|------|------| | id | varchar | 课程ID | 主键 | "C001" | | courseName | varchar | 课程名称 | 必填 | "高等数学" | | credit | int | 学分 | 必填 | 4 | | teacherId | varchar | 授课教师ID | 外键(teachers.id) | "T001" | | semester | varchar | 学期 | 格式"YYYY-N" | "2023-1" | | type | enum | 课程类型 | "必修"或"选修" | "必修" | | prerequisite | varchar | 先修课程ID | 可选,外键(courses.id) | "C003" | ## 4. 学生表 (students) | 字段名 | 类型 | 描述 | 约束 | 示例 | |--------|------|------|------|------| | id | varchar | 学号 | 主键 | "S20230101" | | name | varchar | 学生姓名 | 必填 | "王强" | | gender | enum | 性别 | "男"或"女" | "男" | | birthDate | date | 出生日期 | 必填 | date("2005-01-15") | | enrollmentDate | date | 入学日期 | 必填 | date("2023-8-1") | | classId | varchar | 班级ID | 外键(classes.id) | "202301" | | phone | varchar | 联系电话 | 必填 | "13812345678" | | email | varchar | 电子邮箱 | 必填 | "20230101@school.edu.cn" | | emergencyContact | varchar | 紧急联系人电话 | 必填 | "13876543210" | | address | varchar | 家庭住址 | 必填 | "北京市海淀区中关村大街1栋101室" | | height | int | 身高(cm) | 必填 | 175 | | weight | int | 体重(kg) | 必填 | 65 | | healthStatus | enum | 健康状况 | 必填，"良好"或"一般"或"较差" | "良好" | ## 5. 成绩表 (scores) | 字段名 | 类型 | 描述 | 约束 | 示例 | |--------|------|------|------|------| | id | varchar | 成绩记录ID | 主键 | "S20230101C001" | | studentId | varchar | 学生ID | 外键(students.id) | "S20230101" | | courseId | varchar | 课程ID | 外键(courses.id) | "C001" | | score | int | 综合成绩 | 0-100 | 85 | | examDate | date | 考试日期 | 必填 | date("2024-5-20") | | usualScore | int | 平时成绩 | 0-100 | 90 | | finalScore | int | 期末成绩 | 0-100 | 80 | ### 补考成绩记录说明 补考记录在_id后添加"_M"后缀，如"S20230101C001_M" ## 表关系说明 1. **一对多关系**: - 一个班级(classes)对应多个学生(students) - 一个教师(teachers)可以教授多门课程(courses) - 一个学生(students)有多条成绩记录(scores) 2. **外键约束**: - students.classId → classes.id - courses.teacherId → teachers.id - scores.studentId → students.id - scores.courseId → courses.id - classes.headTeacherId → teachers.id
View Code
mysql表结构，参考文章：https://www.cnblogs.com/xiao987334176/p/18826422
将智能体，添加助手
打开助手，选择MCP
提一个问题，李华的老师是谁
效果如下：
![洛谷 P11345 [KTSC 2023 R2] 基地简化 题解](http://www.itfaba.com/wp-content/themes/kemi/timthumb.php?src=http://www.itfaba.com/wp-content/themes/kemi/img/random/2.jpg&w=218&h=124&zc=1)
