python开发Streamable HTTP MCP应用

一、概述

使用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 FastMCPmcp = FastMCP(\"Demo 🚀\")@mcp.tool()def add(a: int, b: int) -> int: \"\"\"Add two numbers\"\"\" return a + bif __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 jsonimport requestsfrom fastmcp import FastMCPmcp = 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:202INFO: 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是否正确，打开网页

http://ip-api.com/json

结果是正确的，没问题。

三、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 FastMCPfrom mysql.connector import connect, Errorimport osmcp = 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.21ADD . /appRUN 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 fastmcpWORKDIR /appEXPOSE 9000ENTRYPOINT [\"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

 提一个问题，李华的老师是谁

效果如下：