python开发Streamable HTTP MCP应用小结

pip install --upgrade fastmcp

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

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

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

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

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 .

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