SmartBizTrip/mcp_server/mcp_ticket_server.py

# 【启动 S2】独立 MCP 工具服务，监听 8001；供 A2A 子助手调用。
# 本文件负责执行工具，不调用聊天模型来决定下一步；工具名和参数由上游传入。
# 启动顺序：创建 FastMCP → 创建业务服务 → 注册工具 → 转成 FastAPI 应用 → 监听端口。
# 主流程 8：收到工具调用 → 进入对应包装函数 → 业务方法访问数据源/生成模拟结果
# → 返回工具结果 → A2A 执行器交给模型整理 → 主流程 9 返回主助手。

"""
票务 MCP Server 模块
====================
功能：为企业差旅智能体提供票务相关的 MCP 工具（Tool）
  - 高铁票查询 / 预定
  - 航班查询 / 预定
  - 会展门票查询 / 预定

技术要点：
  1. 使用 MySQL 参数化查询（%s 占位符），防止 SQL 注入
  2. 通过 FastMCP 框架将 Python 方法注册为 MCP 工具，供 Agent 调用
  3. 自定义 JSON 编码器处理 date/datetime/Decimal 等特殊类型
"""

import mysql.connector
import json
from datetime import date, datetime, timedelta
from decimal import Decimal
from python_a2a import FastMCP, create_fastapi_app  # MCP 框架：FastMCP 用于定义工具，create_fastapi_app 转为 FastAPI 服务
import uvicorn  # ASGI 服务器，用于运行 FastAPI 应用

from config import Config
from create_logger import logger
from utils.format import DateEncoder, default_encoder

# 加载配置文件中的数据库连接参数
conf = Config()


# ============================================================
# 票务服务类 —— 封装所有票务相关的数据库查询和预定逻辑
# ============================================================
class TicketService:
    """
    票务服务类，负责高铁票、航班、会展门票的查询与预定。
    所有查询方法统一通过 _execute_query 执行参数化 SQL，返回标准 JSON 字符串。
    """

    def __init__(self):
        """初始化方法：建立 MySQL 数据库连接"""
        # 【启动 S2.2 内部】建立并保存连接；后续查询复用 self.conn。
        # 构造函数由服务启动函数调用，一次建立数据库连接并放入 self.conn。
        # self 是业务服务对象；MCP 包装函数接收参数后调用的是这个对象的方法。
        # 成功返回连接对象只说明连接建立，具体表是否存在、查询是否有数据还要等 execute。
        self.conn = mysql.connector.connect(
            host=conf.host,            # 数据库主机地址，如 "localhost" 或 "192.168.1.100"
            user=conf.user,            # 数据库用户名
            password=conf.password,    # 数据库密码
            database=conf.database     # 数据库名称
        )

    def _execute_query(self, sql: str, params: list = None) -> str:
        """
        执行参数化查询的通用方法。

        参数:
            sql:    SQL 语句，使用 %s 作为参数占位符（注意不是 Python 的 % 格式化）
            params: 参数列表，按顺序填充 %s 占位符

        返回:
            JSON 字符串，包含三种可能的结构：
              - {"status": "success", "data": [...]}       查询到数据
              - {"status": "no_data", "message": "..."}    无匹配数据
              - {"status": "error", "message": "..."}      SQL 执行异常

        关键制度要点：
          - dictionary=True：cursor 返回字典而非元组，便于通过列名访问
          - default_encoder：将日期/时间转为字符串，Decimal 转为浮点数
          - DateEncoder：json.dumps 的自定义编码器类
        """
        # 【主流程 8.3，数据库内部步骤 1】创建字典游标，SQL 与参数分别传给 execute。
        # %s 是数据库驱动的参数占位符；参数顺序要与 SQL 中占位符顺序一致。
        try:
            # 创建游标，dictionary=True 使结果以字典列表形式返回
            # 游标是这一次数据库操作的执行入口；dictionary=True 让每行结果按列名取值。
            # 例：{"city": "北京", "fx_date": 日期对象}，比按 results[0][1] 这样的列位置读取更直观。
            cursor = self.conn.cursor(dictionary=True)
            # 执行参数化查询 —— params 中的值会自动转义，防止 SQL 注入
            # SQL 字符串描述查哪些表和条件，参数列表只提供值，二者由驱动组合。
            # 不要把这里的 %s 当成 Python 字符串格式化去替换；当前代码没有将用户输入拼进 SQL 语法。
            cursor.execute(sql, params or [])
            # 【主流程 8.3，数据库内部步骤 2】取回全部匹配行，再关闭此次查询的游标。
            results = cursor.fetchall()
            # 关闭当前游标，self.conn 仍保留供下一次工具调用使用；这里没有关闭整条数据库连接。
            cursor.close()

            # 遍历结果，转换为 JSON 能处理的字符串或数值
            # 【主流程 8.3，数据库内部步骤 3】把日期/时间转为字符串，Decimal 转为 float。
            # 这些值随后才能放入 JSON；转换 Decimal 为 float 可能丢失部分精度。
            for result in results:
                # 外层循环逐行，内层循环逐列。items() 每次给出列名 key 和该列的值 value。
                # 给 result[key] 重新赋值是在替换该列的 Python 表示，没有修改数据库里的原始数据。
                for key, value in result.items():
                    if isinstance(value, (date, datetime, timedelta, Decimal)):
                        result[key] = default_encoder(value)

            # 根据是否有数据返回不同的 JSON 结构
            # 【主流程 8.4】把 success/no_data 和数据序列化为 JSON 文本交回工具调用方。
            # 异常分支返回 error 文本；工具成功返回文本本身不代表查询有结果。
            return json.dumps(
                {"status": "success", "data": results} if results
                else {"status": "no_data", "message": "未找到票务数据，请确认查询条件。"},
                cls=DateEncoder, ensure_ascii=False  # ensure_ascii=False 保证中文正常显示
            )
        # 查询、取值或序列化出错时进入异常分支，返回含 status="error" 的 JSON 文本。
        # 由于错误被转换为普通返回值，上层还需要检查内容，不能只根据“函数正常返回”判断业务成功。
        except Exception as e:  # noqa: BLE001
            # 记录错误日志并返回错误 JSON
            logger.error(f"票务查询错误: {str(e)}")
            return json.dumps({"status": "error", "message": str(e)}, ensure_ascii=False)

    def query_train(self, departure_city: str, arrival_city: str, date: str, seat_type: str = None) -> str:
        """
        查询高铁票。

        参数:
            departure_city: 出发城市，如 "北京"
            arrival_city:   到达城市，如 "上海"
            date:           出发日期，格式 "YYYY-MM-DD"
            seat_type:      座位类型（可选），如 "一等座"、"二等座"

        返回:
            JSON 字符串，包含车次、时间、价格、余票等信息

        SQL 说明：
          - WHERE 条件使用 %s 占位符，参数通过 params 列表传入
          - 如果指定了 seat_type，动态追加 AND 条件
          - DATE(departure_time) 提取日期部分进行匹配
        """
        # 【主流程 8.2】将已传入的参数组织成固定 SQL 和参数列表。
        # 调用者：同名 MCP 包装函数。输入：出发城市、到达城市、日期；输出：查询结果的 JSON 字符串。
        # 数据源：MySQL 的 train_tickets 表，SQL 写在代码中，由参数决定筛选值。
        # 例如 seat_type="二等座" 才追加座席条件；未提供时保留该日各座席类型。
        sql = ("SELECT id, departure_city, arrival_city, departure_time, arrival_time, "
               "train_number, seat_type, price, remaining_seats FROM train_tickets "
               "WHERE departure_city = %s AND arrival_city = %s AND DATE(departure_time) = %s")
        # 参数列表的顺序与 SQL 占位符顺序一致；列表中的日期仍按传入值交给数据库比较。
        params = [departure_city, arrival_city, date]
        # 可选条件有值才追加；同时追加参数，保证条件与参数一一对应。
        if seat_type:
            sql += " AND seat_type = %s"
            # 必须在追加可选 SQL 条件时同时追加对应值，否则占位符数量与参数数量会不一致。
            params.append(seat_type)
        # 【主流程 8.3】进入通用查询方法；结果沿当前调用栈原路返回 MCP 包装函数。
        return self._execute_query(sql, params)

    def query_flight(self, departure_city: str, arrival_city: str, date: str, cabin_type: str = None) -> str:
        """
        查询航班。

        参数:
            departure_city: 出发城市
            arrival_city:   到达城市
            date:           出发日期，格式 "YYYY-MM-DD"
            cabin_type:     舱位类型（可选），如 "经济舱"、"商务舱"

        返回:
            JSON 字符串，包含航班号、时间、价格、余票等信息
        """
        # 【主流程 8.2】将已传入的参数组织成固定 SQL 和参数列表。
        # 调用者：同名 MCP 包装函数。输入：出发城市、到达城市、日期；输出：查询结果的 JSON 字符串。
        # 数据源：MySQL 的 flight_tickets 表，SQL 写在代码中，由参数决定筛选值。
        # 例如 cabin_type="经济舱" 才追加舱位条件；未提供时查询所有符合基础条件的舱位。
        sql = ("SELECT id, departure_city, arrival_city, departure_time, arrival_time, "
               "flight_number, cabin_type, price, remaining_seats FROM flight_tickets "
               "WHERE departure_city = %s AND arrival_city = %s AND DATE(departure_time) = %s")
        # 参数列表的顺序与 SQL 占位符顺序一致；列表中的日期仍按传入值交给数据库比较。
        params = [departure_city, arrival_city, date]
        # 可选条件有值才追加；同时追加参数，保证条件与参数一一对应。
        if cabin_type:
            sql += " AND cabin_type = %s"
            # 必须在追加可选 SQL 条件时同时追加对应值，否则占位符数量与参数数量会不一致。
            params.append(cabin_type)
        # 【主流程 8.3】进入通用查询方法；结果沿当前调用栈原路返回 MCP 包装函数。
        return self._execute_query(sql, params)

    def query_event(self, city: str, event_name: str, date: str, ticket_type: str = None) -> str:
        """
        查询会展活动/会展票。

        参数:
            city:        举办城市，如 "北京"
            event_name:  活动名称，如 "成都企业数字化转型峰会"
            date:        举办日期，格式 "YYYY-MM-DD"
            ticket_type: 票类型（可选），如 "VIP"、"普通票"

        返回:
            JSON 字符串，包含活动名称、场馆、时间、票价、余票等信息
        """
        # 【主流程 8.2】将已传入的参数组织成固定 SQL 和参数列表。
        # 调用者：同名 MCP 包装函数。输入：举办城市、活动名、日期；输出：查询结果的 JSON 字符串。
        # 数据源：MySQL 的 event_tickets 表，SQL 写在代码中，由参数决定筛选值。
        # event_name 使用等值匹配；这不是语义检索，也没有使用 LIKE 做模糊匹配。
        sql = ("SELECT id, event_name, city, venue, start_time, end_time, "
               "ticket_type, price, remaining_seats FROM event_tickets "
               "WHERE city = %s AND event_name = %s AND DATE(start_time) = %s")
        # 参数列表的顺序与 SQL 占位符顺序一致；列表中的日期仍按传入值交给数据库比较。
        params = [city, event_name, date]
        # 可选条件有值才追加；同时追加参数，保证条件与参数一一对应。
        if ticket_type:
            sql += " AND ticket_type = %s"
            # 必须在追加可选 SQL 条件时同时追加对应值，否则占位符数量与参数数量会不一致。
            params.append(ticket_type)
        # 【主流程 8.3】进入通用查询方法；结果沿当前调用栈原路返回 MCP 包装函数。
        return self._execute_query(sql, params)

    # --------------------------------------------------------
    # 预定方法（演示环境：仅模拟返回，未实际写入数据库）
    # --------------------------------------------------------

    def order_train(self, departure_date: str, train_number: str, seat_type: str, number: int) -> str:
        """模拟高铁票预定，返回预定成功信息"""
        # 【主流程 8.2：模拟预订分支】只构造演示文本（部分方法同时记日志）。
        # 这里没有写订单、扣库存、支付、出票或生成保单的动作。
        # 参数用于填充下方 f-string，例如日期、名称、类型和数量会直接显示在演示回复里。
        # 返回字符串后本方法立即结束；它没有再调用查询方法核对库存，也没有生成可持久化的订单对象。
        return f"高铁票预定模拟成功，未实际下单、未创建订单、未扣减库存、未出票或生成保单。日期：{departure_date}，车次：{train_number}，座位：{seat_type}，数量：{number}张。"

    def order_flight(self, departure_date: str, flight_number: str, cabin_type: str, number: int) -> str:
        """模拟航班预定，返回预定成功信息"""
        # 【主流程 8.2：模拟预订分支】只构造演示文本（部分方法同时记日志）。
        # 这里没有写订单、扣库存、支付、出票或生成保单的动作。
        # 参数用于填充下方 f-string，例如日期、名称、类型和数量会直接显示在演示回复里。
        # 返回字符串后本方法立即结束；它没有再调用查询方法核对库存，也没有生成可持久化的订单对象。
        return f"航班预定模拟成功，未实际下单、未创建订单、未扣减库存、未出票或生成保单。日期：{departure_date}，航班号：{flight_number}，舱位：{cabin_type}，数量：{number}张。"

    def order_event(self, start_date: str, event_name: str, venue: str, ticket_type: str, number: int) -> str:
        """模拟会展门票预定，返回预定成功信息"""
        # 【主流程 8.2：模拟预订分支】只构造演示文本（部分方法同时记日志）。
        # 这里没有写订单、扣库存、支付、出票或生成保单的动作。
        # 参数用于填充下方 f-string，例如日期、名称、类型和数量会直接显示在演示回复里。
        # 返回字符串后本方法立即结束；它没有再调用查询方法核对库存，也没有生成可持久化的订单对象。
        return f"会展票预定模拟成功，未实际下单、未创建订单、未扣减库存、未出票或生成保单。日期：{start_date}，活动名称：{event_name}，场馆：{venue}，票类型：{ticket_type}，数量：{number}张。"


def create_ticket_mcp_server():
    # 【启动 S2.1】创建工具注册表。下面的 @tool 在定义包装函数时完成注册。
    # 注册时不会运行包装函数体；只有收到该工具的调用请求才执行函数体。
    # 这是服务启动函数。内部定义的工具包装函数会记住当前的 service 对象，后续每次调用可继续使用它。
    # 这叫闭包：包装函数虽然在外层函数内部定义，却能读取外层的 service。
    # FastMCP 注册这些函数后，HTTP 请求按工具名选择其中一个，不按源代码排列顺序逐个执行。
    ticket_mcp = FastMCP(
        name="TicketTools",
    )

    # 【启动 S2.2】创建可被所有工具包装函数引用的业务服务，并建立 MySQL 连接。
    service = TicketService()

    # 注册查询工具
    @ticket_mcp.tool(
        name="query_train",
        description="查询高铁票信息，参数：departure_city(出发城市), arrival_city(到达城市), date(日期), seat_type(座位类型，可选)"
    )
    def query_train(departure_city: str, arrival_city: str, date: str, seat_type: str = None) -> str:
        # 【主流程 8.1】客户端选中本工具时才进入这里；把命名参数转交业务方法。
        # 调用者：MCP 框架按工具名 query_train 分派到这里。
        # 输入：函数签名中的命名参数；输出：业务方法返回的字符串，再由 MCP 框架封装为工具响应。
        # 同名的 TicketService.query_train 是业务方法；当前嵌套函数只是对外暴露的工具入口。
        return service.query_train(departure_city, arrival_city, date, seat_type)

    @ticket_mcp.tool(
        name="query_flight",
        description="查询航班信息，参数：departure_city(出发城市), arrival_city(到达城市), date(日期), cabin_type(舱位类型，可选)"
    )
    def query_flight(departure_city: str, arrival_city: str, date: str, cabin_type: str = None) -> str:
        # 【主流程 8.1】客户端选中本工具时才进入这里；把命名参数转交业务方法。
        # 调用者：MCP 框架按工具名 query_flight 分派到这里。
        # 输入：函数签名中的命名参数；输出：业务方法返回的字符串，再由 MCP 框架封装为工具响应。
        # 同名的 TicketService.query_flight 是业务方法；当前嵌套函数只是对外暴露的工具入口。
        return service.query_flight(departure_city, arrival_city, date, cabin_type)

    @ticket_mcp.tool(
        name="query_event",
        description="查询会展门票信息，参数：city(城市), event_name(活动名称), date(日期), ticket_type(票类型，可选)"
    )
    def query_event(city: str, event_name: str, date: str, ticket_type: str = None) -> str:
        # 【主流程 8.1】客户端选中本工具时才进入这里；把命名参数转交业务方法。
        # 调用者：MCP 框架按工具名 query_event 分派到这里。
        # 输入：函数签名中的命名参数；输出：业务方法返回的字符串，再由 MCP 框架封装为工具响应。
        # 同名的 TicketService.query_event 是业务方法；当前嵌套函数只是对外暴露的工具入口。
        return service.query_event(city, event_name, date, ticket_type)

    # 注册预定工具
    @ticket_mcp.tool(
        name="order_train",
        description="仅模拟演示，不实际下单、不创建订单、不扣减库存、不出票或生成保单。根据车次号、日期、座位类型和数量预定高铁票。参数：train_number(车次号，如G123), departure_date(出发日期，格式YYYY-MM-DD), seat_type(座位类型，如二等座/一等座/商务座), number(订票数量)"
    )
    def order_train(departure_date: str, train_number: str, seat_type: str, number: int) -> str:
        # 【主流程 8.1】客户端选中本工具时才进入这里；把命名参数转交业务方法。
        # 调用者：MCP 框架按工具名 order_train 分派到这里。
        # 输入：函数签名中的命名参数；输出：业务方法返回的字符串，再由 MCP 框架封装为工具响应。
        # 同名的 TicketService.order_train 是业务方法；当前嵌套函数只是对外暴露的工具入口。
        return service.order_train(departure_date, train_number, seat_type, number)

    @ticket_mcp.tool(
        name="order_flight",
        description="仅模拟演示，不实际下单、不创建订单、不扣减库存、不出票或生成保单。根据航班号、日期、舱位类型和数量预定航班。参数：flight_number(航班号，如CA4101), departure_date(出发日期，格式YYYY-MM-DD), cabin_type(舱位类型，如经济舱/商务舱/头等舱), number(订票数量)"
    )
    def order_flight(departure_date: str, flight_number: str, cabin_type: str, number: int) -> str:
        # 【主流程 8.1】客户端选中本工具时才进入这里；把命名参数转交业务方法。
        # 调用者：MCP 框架按工具名 order_flight 分派到这里。
        # 输入：函数签名中的命名参数；输出：业务方法返回的字符串，再由 MCP 框架封装为工具响应。
        # 同名的 TicketService.order_flight 是业务方法；当前嵌套函数只是对外暴露的工具入口。
        return service.order_flight(departure_date, flight_number, cabin_type, number)

    @ticket_mcp.tool(
        name="order_event",
        description="仅模拟演示，不实际下单、不创建订单、不扣减库存、不出票或生成保单。根据活动名称、日期、场馆、票类型和数量预定会展门票。参数：event_name(活动名称), start_date(举办日期，格式YYYY-MM-DD), venue(会展场馆名), ticket_type(票类型，如VIP/普通票), number(订票数量)"
    )
    def order_event(start_date: str, event_name: str, venue: str, ticket_type: str, number: int) -> str:
        # 【主流程 8.1】客户端选中本工具时才进入这里；把命名参数转交业务方法。
        # 调用者：MCP 框架按工具名 order_event 分派到这里。
        # 输入：函数签名中的命名参数；输出：业务方法返回的字符串，再由 MCP 框架封装为工具响应。
        # 同名的 TicketService.order_event 是业务方法；当前嵌套函数只是对外暴露的工具入口。
        return service.order_event(start_date, event_name, venue, ticket_type, number)

    logger.info("=== 票务MCP服务器信息 ===")
    logger.info(f"名称: {ticket_mcp.name}")

    try:
        print("服务器已启动，请访问 http://127.0.0.1:8001/mcp")
        # 【启动 S2.3】工具注册结束后构建 HTTP 应用；下一行 uvicorn.run 开始监听并等待请求。
        ticket_mcp_server = create_fastapi_app(ticket_mcp)
        uvicorn.run(ticket_mcp_server, host="0.0.0.0", port=8001)
    except Exception as e:
        print(f"服务器启动失败: {e}")


if __name__ == '__main__':
    # service = TicketService()
    # train = service.query_train("北京", "成都", "2026-05-01", "二等座")
    # print(train)
    # flight = service.query_flight("北京", "成都", "2026-05-01", "头等舱")
    # print(flight)
    create_ticket_mcp_server()