目录

• 一、概述
• 二、Streamable HTTP MCP应用官方demo
• 三、SSE转换为Streamable HTTP代码改造

一、概述

使用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.jb51.net/javascript/340794xx1.htm

这里就不再重复了

二、Streamable HTTP MCP应用官方demo

server.py

from fastmcp import FastMCPmcp = FastMCP(“Demo &x1f680;”)@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 &x1f680;”)@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"&8230; 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版本是最新的，由于新版本，增加了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.jb51.net/server/321891t37.htm

把代码拷贝过来，只需要修改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

mysql表结构，参考文章：https://www.jb51.net/database/3415585jf.htm

将智能体，添加助手

打开助手，选择MCP

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

效果如下：

到此这篇关于python开发StreamableHTTPMCP应用的文章就介绍到这了,更多相关python开发StreamableHTTPMCP应用内容请搜索风君子博客以前的文章或继续浏览下面的相关文章希望大家以后多多支持风君子博客！

无论兄弟们可能感兴趣的文章:

• python开发Streamable?HTTP?MCP应用
• python中httpx库的详细使用技巧及案例详解
• Python怎样实现HTTPecho服务器
• python清华pip镜像源报HTTPerror403错误解决办法
• Python中的http.server库用法详细介绍