python开发Streamable HTTP MCP应用

pip install --upgrade fastmcp

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")

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)

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")

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 .

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

docker-compose up -d

使用中文回复。当用户提问中涉及学生、教师、成绩、班级、课程等实体时，需要使用 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

CREATE TABLE `classes` (  `id` varchar(255) CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci NOT NULL COMMENT '班级ID，示例：202301',  `className` varchar(255) CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci NOT NULL COMMENT '班级名称，示例：2023级计算机1班',  `grade` int NOT NULL COMMENT '年级，示例：2023',  `headTeacherId` varchar(255) CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci NOT NULL COMMENT '班主任ID，外键(teachers.id)，示例：T003',  `classroom` varchar(255) CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci NOT NULL COMMENT '教室位置，示例：1号楼302',  `studentCount` int NOT NULL COMMENT '学生人数，示例：35',  `remark` varchar(255) CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci DEFAULT NULL COMMENT '备注信息，示例：市级优秀班集体',  PRIMARY KEY (`id`),  KEY `headTeacherId` (`headTeacherId`),  CONSTRAINT `headTeacherId` FOREIGN KEY (`headTeacherId`) REFERENCES `teachers` (`id`) ON DELETE CASCADE ON UPDATE CASCADE) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci COMMENT='班级表';

CREATE TABLE `courses` (  `id` varchar(255) CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci NOT NULL COMMENT '课程ID，示例：C001',  `courseName` varchar(255) CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci NOT NULL COMMENT '课程名称，示例：高等数学',  `credit` int NOT NULL COMMENT '学分，示例：4',  `teacherId` varchar(255) CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci NOT NULL COMMENT '授课教师ID，外键(teachers.id)，示例：T001',  `semester` varchar(255) CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci DEFAULT NULL COMMENT '学期，格式"YYYY-N"，示例：2023-1',  `type` enum('必修','选修') CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci NOT NULL DEFAULT '选修' COMMENT '课程类型，"必修"或"选修"，示例：选修',  `prerequisite` varchar(255) COLLATE utf8mb4_unicode_ci DEFAULT NULL COMMENT '先修课程ID，可选，外键(courses.id)，示例：C003',  PRIMARY KEY (`id`),  KEY `teacherId` (`teacherId`),  CONSTRAINT `teacherId` FOREIGN KEY (`teacherId`) REFERENCES `teachers` (`id`) ON DELETE CASCADE ON UPDATE CASCADE) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci COMMENT='课程表';

CREATE TABLE `scores` (  `id` varchar(255) CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci NOT NULL COMMENT '成绩记录ID，示例：S20230101C001',  `studentId` varchar(255) CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci NOT NULL COMMENT '学生ID，外键(students.id)，示例：S20230101',  `courseId` varchar(255) CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci NOT NULL COMMENT '课程ID，外键(courses.id)，示例：C001',  `score` int NOT NULL COMMENT '综合成绩，0-100，示例：85',  `examDate` date NOT NULL COMMENT '考试日期，示例：2024-5-20',  `usualScore` int DEFAULT '0' COMMENT '平时成绩，0-100，示例：90',  `finalScore` int DEFAULT '0' COMMENT '期末成绩，0-100，示例：80',  PRIMARY KEY (`id`),  KEY `studentId` (`studentId`),  KEY `courseId` (`courseId`),  CONSTRAINT `courseId` FOREIGN KEY (`courseId`) REFERENCES `courses` (`id`) ON DELETE CASCADE ON UPDATE CASCADE,  CONSTRAINT `studentId` FOREIGN KEY (`studentId`) REFERENCES `students` (`id`) ON DELETE CASCADE ON UPDATE CASCADE) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci COMMENT='成绩表';

CREATE TABLE `students` (  `id` varchar(255) CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci NOT NULL COMMENT '学号，示例：S20230101',  `name` varchar(255) CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci NOT NULL COMMENT '学生姓名，示例：王强',  `gender` enum('男','女') CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci NOT NULL DEFAULT '男' COMMENT '性别，"男"或"女"，示例：男',  `birthDate` datetime NOT NULL COMMENT '出生日期，示例：2005-01-15',  `classId` varchar(255) CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci NOT NULL COMMENT '班级ID，外键(classes.id)，示例：202301',  `phone` varchar(255) CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci NOT NULL COMMENT '联系电话，示例：13812345678',  `email` varchar(255) CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci NOT NULL COMMENT '电子邮箱，示例：20230101@school.edu.cn',  `emergencyContact` varchar(255) CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci NOT NULL COMMENT '紧急联系人电话，示例：13876543210',  `address` varchar(255) CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci NOT NULL COMMENT '家庭住址，示例：北京市海淀区中关村大街1栋101室',  `height` int NOT NULL COMMENT '身高(cm)，示例：175',  `weight` int NOT NULL COMMENT '体重(kg)，示例：65',  `healthStatus` enum('良好‌','一般‌','较差') CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci NOT NULL DEFAULT '良好‌' COMMENT '健康状况，示例：良好',  PRIMARY KEY (`id`),  KEY `classId` (`classId`),  CONSTRAINT `classId` FOREIGN KEY (`classId`) REFERENCES `classes` (`id`) ON DELETE CASCADE ON UPDATE CASCADE) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci COMMENT='学生表';

CREATE TABLE `teachers` (  `id` varchar(255) CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci NOT NULL COMMENT '教师ID，示例：T001',  `name` varchar(255) CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci NOT NULL COMMENT '教师姓名，示例：张建国',  `gender` enum('男','女') CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci NOT NULL DEFAULT '男' COMMENT '性别，"男"或"女"，示例：男',  `subject` varchar(255) CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci NOT NULL COMMENT '教授科目，示例：数学',  `title` varchar(255) CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci NOT NULL COMMENT '职称，示例：教授',  `phone` varchar(255) CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci NOT NULL COMMENT '联系电话，示例：13812345678',  `office` varchar(255) CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci NOT NULL COMMENT '办公室位置，示例：博学楼301',  `wechat` varchar(255) CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci DEFAULT NULL COMMENT '微信，示例：lily_teacher',  `isHeadTeacher` enum('true','false') CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci DEFAULT 'false' COMMENT '是否为班主任，示例：true',  PRIMARY KEY (`id`,`office`) USING BTREE,  KEY `id` (`id`)) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci COMMENT='教师表';