python開發(fā)Streamable?HTTP?MCP應(yīng)用小結(jié)
更新時(shí)間：2025年05月16日 08:47:50 作者：肖祥
這篇文章主要介紹了python開發(fā)Streamable?HTTP?MCP應(yīng)用的記錄,本文給大家介紹的非常詳細(xì),對(duì)大家的學(xué)習(xí)或工作具有一定的參考借鑒價(jià)值,需要的朋友參考下吧
一、概述
使用python開發(fā)，最好的框架是fastmcp，github連接：https://github.com/jlowin/fastmcp
2025 年 5 月 9 日，fastmcp發(fā)布v2.3.0版本，正式支持Streamable HTTP
終于等到官方支持了！
注意：2.3.0版本有bug，目前最新版本已經(jīng)修復(fù)了
升級(jí)到最新版本
pip install --upgrade fastmcp
關(guān)于Streamable HTTP的介紹，請(qǐng)參考鏈接：http://m.dhdzp.com/javascript/340794xx1.htm
這里就不再重復(fù)了
二、Streamable HTTP MCP應(yīng)用官方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行代碼，就簡(jiǎn)單實(shí)現(xiàn)了Streamable HTTP MCP應(yīng)用。
獲取服務(wù)器公網(wǎng)ip
通過一個(gè)在實(shí)際生產(chǎn)環(huán)境中，使用的功能，來演示一個(gè)Streamable HTTP MCP應(yīng)用
server.py
import json
import requests
from fastmcp import FastMCP
mcp = FastMCP("Demo ??")
@mcp.tool()
def get_public_ip_address() -> str:
"""
獲取服務(wù)器公網(wǎng) IP 地址
返回:
str: 當(dāng)前網(wǎng)絡(luò)的公網(wǎng) IP 地址
"""
try:
response = requests.get("http://ip-api.com/json")
response.raise_for_status() # 檢查 HTTP 請(qǐng)求是否成功
content = json.loads(response.text)
return content.get("query", "Unknown IP") # 提供默認(rèn)值以防字段缺失
except requests.RequestException as e:
print(f"請(qǐng)求錯(cuò)誤: {e}")
return "Request Failed"
except json.JSONDecodeError as e:
print(f"JSON 解碼錯(cuò)誤: {e}")
return "Invalid Response"
if __name__ == "__main__":
# mcp.run()
mcp.run(transport="streamable-http",
host="0.0.0.0", port=9000, path="/mcp")運(yùn)行代碼
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)
請(qǐng)確保Cherry Studio版本是最新的，因?yàn)樾掳姹?，增加了Streamable HTTP支持
添加mcp服務(wù)器
名稱：public_ip_address_mcp
類型：Streamable HTTP
url：http://localhost:9000/mcp
添加成功后，點(diǎn)擊工具，可以看到工具方法
創(chuàng)建
添加
選擇助手
選擇
提問公網(wǎng)ip，效果如下：
驗(yàn)證一下公網(wǎng)ip是否正確，打開網(wǎng)頁
結(jié)果是正確的，沒問題。
三、SSE轉(zhuǎn)換為Streamable HTTP代碼改造
在前面的文章中，寫的mysql應(yīng)用是SSE模式，鏈接：http://m.dhdzp.com/server/321891t37.htm
把代碼拷貝過來，只需要修改2行代碼，就可以無縫轉(zhuǎn)換為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應(yīng)用，完整代碼如下：
server.py
from fastmcp import FastMCP
from mysql.connector import connect, Error
import os
mcp = FastMCP("operateMysql")
def get_db_config():
"""從環(huán)境變量獲取數(shù)據(jù)庫配置信息
返回:
dict: 包含數(shù)據(jù)庫連接所需的配置信息
- host: 數(shù)據(jù)庫主機(jī)地址
- port: 數(shù)據(jù)庫端口
- user: 數(shù)據(jù)庫用戶名
- password: 數(shù)據(jù)庫密碼
- database: 數(shù)據(jù)庫名稱
異常:
ValueError: 當(dāng)必需的配置信息缺失時(shí)拋出
"""
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("缺少必需的數(shù)據(jù)庫配置")
return config
@mcp.tool()
def execute_sql(query: str) -> list:
"""執(zhí)行SQL查詢語句
參數(shù):
query (str): 要執(zhí)行的SQL語句，支持多條語句以分號(hào)分隔
返回:
list: 包含查詢結(jié)果的TextContent列表
- 對(duì)于SELECT查詢：返回CSV格式的結(jié)果，包含列名和數(shù)據(jù)
- 對(duì)于SHOW TABLES：返回?cái)?shù)據(jù)庫中的所有表名
- 對(duì)于其他查詢：返回執(zhí)行狀態(tài)和影響行數(shù)
- 多條語句的結(jié)果以"---"分隔
異常:
Error: 當(dāng)數(shù)據(jù)庫連接或查詢執(zhí)行失敗時(shí)拋出
"""
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)
# 檢查語句是否返回了結(jié)果集 (SELECT, SHOW, EXPLAIN, etc.)
if cursor.description:
columns = [desc[0] for desc in cursor.description]
rows = cursor.fetchall()
# 將每一行的數(shù)據(jù)轉(zhuǎn)換為字符串，特殊處理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))
# 將列名和數(shù)據(jù)合并為CSV格式
results.append(
"\n".join([",".join(columns)] + formatted_rows)
)
# 如果語句沒有返回結(jié)果集 (INSERT, UPDATE, DELETE, etc.)
else:
conn.commit() # 只有在非查詢語句時(shí)才提交
results.append(f"查詢執(zhí)行成功。影響行數(shù): {cursor.rowcount}")
except Error as stmt_error:
# 單條語句執(zhí)行出錯(cuò)時(shí)，記錄錯(cuò)誤并繼續(xù)執(zhí)行
results.append(
f"執(zhí)行語句 '{statement}' 出錯(cuò): {str(stmt_error)}"
)
# 可以在這里選擇是否繼續(xù)執(zhí)行后續(xù)語句，目前是繼續(xù)
return ["\n---\n".join(results)]
except Error as e:
print(f"執(zhí)行SQL '{query}' 時(shí)出錯(cuò): {e}")
return [f"執(zhí)行查詢時(shí)出錯(cuò): {str(e)}"]
@mcp.tool()
def get_table_name(text: str) -> list:
"""根據(jù)表的中文注釋搜索數(shù)據(jù)庫中的表名
參數(shù):
text (str): 要搜索的表中文注釋關(guān)鍵詞
返回:
list: 包含查詢結(jié)果的TextContent列表
- 返回匹配的表名、數(shù)據(jù)庫名和表注釋信息
- 結(jié)果以CSV格式返回，包含列名和數(shù)據(jù)
"""
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:
"""獲取指定表的字段結(jié)構(gòu)信息
參數(shù):
text (str): 要查詢的表名，多個(gè)表名以逗號(hào)分隔
返回:
list: 包含查詢結(jié)果的列表
- 返回表的字段名、字段注釋等信息
- 結(jié)果按表名和字段順序排序
- 結(jié)果以CSV格式返回，包含列名和數(shù)據(jù)
"""
config = get_db_config()
# 將輸入的表名按逗號(hào)分割成列表
table_names = [name.strip() for name in text.split(",")]
# 構(gòu)建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:
"""
獲取當(dāng)前mysql服務(wù)器InnoDB 的行級(jí)鎖
返回:
list: 包含查詢結(jié)果的TextContent列表
"""
sql = """SELECT
p2.`HOST` AS 被阻塞方host,
p2.`USER` AS 被阻塞方用戶,
r.trx_id AS 被阻塞方事務(wù)id,
r.trx_mysql_thread_id AS 被阻塞方線程號(hào),
TIMESTAMPDIFF(SECOND, r.trx_wait_started, CURRENT_TIMESTAMP) AS 等待時(shí)間,
r.trx_query AS 被阻塞的查詢,
l.OBJECT_NAME AS 阻塞方鎖住的表,
m.LOCK_MODE AS 被阻塞方的鎖模式,
m.LOCK_TYPE AS '被阻塞方的鎖類型(表鎖還是行鎖)',
m.INDEX_NAME AS 被阻塞方鎖住的索引,
m.OBJECT_SCHEMA AS 被阻塞方鎖對(duì)象的數(shù)據(jù)庫名,
m.OBJECT_NAME AS 被阻塞方鎖對(duì)象的表名,
m.LOCK_DATA AS 被阻塞方事務(wù)鎖定記錄的主鍵值,
p.`HOST` AS 阻塞方主機(jī),
p.`USER` AS 阻塞方用戶,
b.trx_id AS 阻塞方事務(wù)id,
b.trx_mysql_thread_id AS 阻塞方線程號(hào),
b.trx_query AS 阻塞方查詢,
l.LOCK_MODE AS 阻塞方的鎖模式,
l.LOCK_TYPE AS '阻塞方的鎖類型(表鎖還是行鎖)',
l.INDEX_NAME AS 阻塞方鎖住的索引,
l.OBJECT_SCHEMA AS 阻塞方鎖對(duì)象的數(shù)據(jù)庫名,
l.OBJECT_NAME AS 阻塞方鎖對(duì)象的表名,
l.LOCK_DATA AS 阻塞方事務(wù)鎖定記錄的主鍵值,
IF(p.COMMAND = 'Sleep', CONCAT(p.TIME, ' 秒'), 0) AS 阻塞方事務(wù)空閑的時(shí)間
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 等待時(shí)間 DESC;"""
return execute_sql(sql)
if __name__ == "__main__":
mcp.run(transport="streamable-http",
host="0.0.0.0", port=9000, path="/mcp")修改Dockerfile，升級(jí)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啟動(dòng)
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相關(guān)環(huán)境變量
運(yùn)行
docker-compose up -d
Cherry Studio測(cè)試
添加MCP服務(wù)器
添加智能體mysql8
提示詞和以前一樣
使用中文回復(fù)。
當(dāng)用戶提問中涉及學(xué)生、教師、成績(jī)、班級(jí)、課程等實(shí)體時(shí)，需要使用 MySQL MCP 進(jìn)行數(shù)據(jù)查詢和操作，表結(jié)構(gòu)說明如下：
# 學(xué)生管理系統(tǒng)數(shù)據(jù)庫表結(jié)構(gòu)說明
## 1. 教師表 (teachers)
| 字段名 | 類型 | 描述 | 約束 | 示例 |
|--------|------|------|------|------|
| id | varchar | 教師ID | 主鍵 | "T001" |
| name | varchar | 教師姓名 | 必填 | "張建國" |
| gender | enum | 性別 | "男"或"女" | "男" |
| subject | varchar | 教授科目 | 必填 | "數(shù)學(xué)" |
| title | varchar | 職稱 | 必填 | "教授" |
| phone | varchar | 聯(lián)系電話 | 必填 | "13812345678" |
| office | varchar | 辦公室位置 | 必填 | "博學(xué)樓301" |
| wechat | varchar | 微信(可選) | 可選 | "lily_teacher" |
| isHeadTeacher | enum | 是否為班主任，"true"或"false" | 可選 | true |
## 2. 班級(jí)表 (classes)
| 字段名 | 類型 | 描述 | 約束 | 示例 |
|--------|------|------|------|------|
| id | varchar | 班級(jí)ID | 主鍵 | "202301" |
| className | varchar | 班級(jí)名稱 | 必填 | "2023級(jí)計(jì)算機(jī)1班" |
| grade | int | 年級(jí) | 必填 | 2023 |
| headTeacherId | varchar | 班主任ID | 外鍵(teachers.id) | "T003" |
| classroom | varchar | 教室位置 | 必填 | "1號(hào)樓302" |
| studentCount | int | 學(xué)生人數(shù) | 必填 | 35 |
| remark | varchar | 備注信息 | 可選 | "市級(jí)優(yōu)秀班集體" |
## 3. 課程表 (courses)
| 字段名 | 類型 | 描述 | 約束 | 示例 |
|--------|------|------|------|------|
| id | varchar | 課程ID | 主鍵 | "C001" |
| courseName | varchar | 課程名稱 | 必填 | "高等數(shù)學(xué)" |
| credit | int | 學(xué)分 | 必填 | 4 |
| teacherId | varchar | 授課教師ID | 外鍵(teachers.id) | "T001" |
| semester | varchar | 學(xué)期 | 格式"YYYY-N" | "2023-1" |
| type | enum | 課程類型 | "必修"或"選修" | "必修" |
| prerequisite | varchar | 先修課程ID | 可選,外鍵(courses.id) | "C003" |
## 4. 學(xué)生表 (students)
| 字段名 | 類型 | 描述 | 約束 | 示例 |
|--------|------|------|------|------|
| id | varchar | 學(xué)號(hào) | 主鍵 | "S20230101" |
| name | varchar | 學(xué)生姓名 | 必填 | "王強(qiáng)" |
| gender | enum | 性別 | "男"或"女" | "男" |
| birthDate | date | 出生日期 | 必填 | date("2005-01-15") |
| enrollmentDate | date | 入學(xué)日期 | 必填 | date("2023-8-1") |
| classId | varchar | 班級(jí)ID | 外鍵(classes.id) | "202301" |
| phone | varchar | 聯(lián)系電話 | 必填 | "13812345678" |
| email | varchar | 電子郵箱 | 必填 | "20230101@school.edu.cn" |
| emergencyContact | varchar | 緊急聯(lián)系人電話 | 必填 | "13876543210" |
| address | varchar | 家庭住址 | 必填 | "北京市海淀區(qū)中關(guān)村大街1棟101室" |
| height | int | 身高(cm) | 必填 | 175 |
| weight | int | 體重(kg) | 必填 | 65 |
| healthStatus | enum | 健康狀況 | 必填，"良好"或"一般?"或"較差" | "良好" |
## 5. 成績(jī)表 (scores)
| 字段名 | 類型 | 描述 | 約束 | 示例 |
|--------|------|------|------|------|
| id | varchar | 成績(jī)記錄ID | 主鍵 | "S20230101C001" |
| studentId | varchar | 學(xué)生ID | 外鍵(students.id) | "S20230101" |
| courseId | varchar | 課程ID | 外鍵(courses.id) | "C001" |
| score | int | 綜合成績(jī) | 0-100 | 85 |
| examDate | date | 考試日期 | 必填 | date("2024-5-20") |
| usualScore | int | 平時(shí)成績(jī) | 0-100 | 90 |
| finalScore | int | 期末成績(jī) | 0-100 | 80 |
### 補(bǔ)考成績(jī)記錄說明
補(bǔ)考記錄在_id后添加"_M"后綴，如"S20230101C001_M"
## 表關(guān)系說明
1. **一對(duì)多關(guān)系**:
- 一個(gè)班級(jí)(classes)對(duì)應(yīng)多個(gè)學(xué)生(students)
- 一個(gè)教師(teachers)可以教授多門課程(courses)
- 一個(gè)學(xué)生(students)有多條成績(jī)記錄(scores)
2. **外鍵約束**:
- students.classId → classes.id
- courses.teacherId → teachers.id
- scores.studentId → students.id
- scores.courseId → courses.id
- classes.headTeacherId → teachers.idmysql表結(jié)構(gòu)，參考文章：http://m.dhdzp.com/database/3415585jf.htm
將智能體，添加助手
打開助手，選擇MCP
提一個(gè)問題，李華的老師是誰
效果如下：
到此這篇關(guān)于python開發(fā)Streamable HTTP MCP應(yīng)用的文章就介紹到這了,更多相關(guān)python開發(fā)Streamable HTTP MCP應(yīng)用內(nèi)容請(qǐng)搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！
您可能感興趣的文章:
- Python快速開發(fā)一個(gè)MCP服務(wù)器的實(shí)現(xiàn)示例
- 使用Python+Bright Data MCP實(shí)時(shí)抓取Google搜索結(jié)果完整教程
- 基于Python從零構(gòu)建一個(gè)MCP服務(wù)器
- 使用Python構(gòu)建MCP服務(wù)器的詳細(xì)配置步驟
- Python使用FastMCP實(shí)現(xiàn)Word文檔與JSON數(shù)據(jù)互轉(zhuǎn)
- Python?FastMCP構(gòu)建MCP服務(wù)端與客戶端的詳細(xì)步驟
- Python?MCPInspector調(diào)試思路詳解
- 通過Python調(diào)用MCP的實(shí)現(xiàn)示例
相關(guān)文章
這篇文章主要介紹了python串口讀取數(shù)據(jù)的實(shí)例,具有很好的參考價(jià)值,希望對(duì)大家有所幫助,如有錯(cuò)誤或未考慮完全的地方,望不吝賜教2023-09-09
使用OpenCV實(shí)現(xiàn)仿射變換—旋轉(zhuǎn)功能
這篇文章主要介紹了在OpenCV里實(shí)現(xiàn)仿射變換——旋轉(zhuǎn)功能，本文通過實(shí)例代碼給大家介紹的非常詳細(xì)，具有一定的參考借鑒價(jià)值,需要的朋友可以參考下2019-08-08
基于Python實(shí)現(xiàn)的購物商城管理系統(tǒng)
這篇文章主要介紹了基于Python實(shí)現(xiàn)的購物商城管理系統(tǒng)，幫助大家更好的理解和學(xué)習(xí)使用python，感興趣的朋友可以了解下2021-04-04
Python3中urlencode和urldecode的用法詳解
今天小編就為大家分享一篇Python3中urlencode和urldecode的用法詳解，具有很好的參考價(jià)值，希望對(duì)大家有所幫助。一起跟隨小編過來看看吧2019-07-07
Python爬取奶茶店數(shù)據(jù)分析哪家最好喝以及性價(jià)比
這篇文章主要介紹了用Python告訴你奶茶哪家最好喝性價(jià)比最高，文中通過示例代碼介紹的非常詳細(xì)，對(duì)大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價(jià)值，需要的朋友們下面隨著小編來一起學(xué)習(xí)吧2022-09-09
Win10操作系統(tǒng)中PyTorch虛擬環(huán)境配置+PyCharm配置
本文主要介紹了Win10操作系統(tǒng)中PyTorch虛擬環(huán)境配置+PyCharm配置，文中通過示例代碼介紹的非常詳細(xì)，具有一定的參考價(jià)值，感興趣的小伙伴們可以參考一下2021-09-09
Python數(shù)據(jù)結(jié)構(gòu)之翻轉(zhuǎn)鏈表
這篇文章主要介紹了Python數(shù)據(jù)結(jié)構(gòu)之翻轉(zhuǎn)鏈表的相關(guān)資料,需要的朋友可以參考下2017-02-02
這篇文章主要為大家詳細(xì)介紹了使用python實(shí)現(xiàn)kNN分類算法，文中示例代碼介紹的非常詳細(xì)，具有一定的參考價(jià)值，感興趣的小伙伴們可以參考一下2019-10-10
這篇文章主要給大家介紹了關(guān)于如何利用Python將圖片中扭曲矩形復(fù)原的相關(guān)資料，文中通過示例代碼介紹的非常詳細(xì)，對(duì)大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價(jià)值，需要的朋友們下面隨著小編來一起學(xué)習(xí)學(xué)習(xí)吧2020-09-09
圖像金字塔是指由一組圖像且不同分別率的子圖集合，它是圖像多尺度表達(dá)的一種，以多分辨率來解釋圖像的結(jié)構(gòu)，主要用于圖像的分割或壓縮。本文主要介紹了圖像金字塔的圖像向下取樣和向上取樣，感興趣的可以了解一下2022-09-09