SmartBizTrip/mcp_server/mcp_trip_server.py

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

import json  # JSON 处理
from datetime import date, datetime, timedelta  # 时间处理
from decimal import Decimal  # 高精度小数类型

import mysql.connector  # MySQL 数据库驱动
import requests  # HTTP 请求，用于调用 Qwen Embedding API
import uvicorn  # ASGI 服务器，用于运行 FastAPI 应用
from pymilvus import Collection, connections  # Milvus 向量数据库客户端
from python_a2a import (  # MCP 框架：FastMCP 用于定义工具，create_fastapi_app 转为 FastAPI 服务
    FastMCP,
    create_fastapi_app,
)
from config import Config  # 项目配置
from create_logger import logger  # 日志模块
from utils.format import DateEncoder, default_encoder  # 日期格式化工具

conf = Config()


def get_embedding(text: str) -> list:
    """调用 Qwen Embedding API 生成文本向量"""
    # 【检索内部步骤 2】把查询文本发给向量模型，得到可与套餐向量比较的一组数值。
    # 查询和入库需使用一致的模型与维度；向量接口不同于聊天模型接口。
    # 调用者：search_meeting_packages_in_milvus。输入是一段用户需求文本，输出是浮点数列表。
    # payload.input 虽是列表，此处只放一条 text，因此最后读取 data[0] 的 embedding。
    import requests
    headers = {
        "Authorization": f"Bearer {conf.embedding_api_key}",
        "Content-Type": "application/json"
    }
    payload = {
        "model": "text-embedding-v3",
        "input": [text],
        "dimensions": conf.embedding_dim
    }
    # POST 将 payload 编码为 JSON，Authorization 使用独立的向量服务凭证。
    # 这里的请求发生在真正检索时，创建 TripService 本身不会提前生成这些向量。
    response = requests.post(conf.embedding_url, headers=headers, json=payload, timeout=30)
    response.raise_for_status()
    return response.json()["data"][0]["embedding"]


def search_meeting_packages_in_milvus(query_text: str, city: str = None, limit: int = 5) -> list:
    """在 Milvus 中进行语义搜索，找到最匹配的会务套餐"""
    # 【检索内部步骤 1】连接并加载已建好的集合；本函数不会自动创建或灌入套餐数据。
    # 输入：query_text 描述套餐需求，city 是可选城市，limit 限制最多返回多少个候选。
    # 输出：普通 Python 字典列表，尚未 json.dumps；外层 query_meeting_package 会再包装状态并序列化。
    # 本函数检索 sql/init_meeting_package_rag.py 事先写入的集合，不从 MySQL 动态同步套餐。
    from pymilvus import connections, Collection
    connections.connect(alias="default", host=conf.milvus_host, port=conf.milvus_port)
    collection = Collection(conf.meeting_package_collection)
    # 把集合准备为可查询状态；Collection(name) 只是取得集合对象，并未创建新套餐数据。
    collection.load()

    # 【检索内部步骤 2 → 3】先求查询向量，再按 COSINE 相似度取前 limit 条。
    # city 有值时额外过滤城市；相似度是检索分数，不能当成事实正确率或预订成功率。
    results = collection.search(
        # search 的 data 是“查询向量列表”；当前只有一个查询向量，所以外层结果通常只有一组命中。
        data=[get_embedding(query_text)],
        # 指定对 embedding 向量字段检索；COSINE 与初始化脚本建立索引时的度量方式保持一致。
        anns_field="embedding",
        param={"metric_type": "COSINE", "params": {"nprobe": 10}},
        limit=limit,
        output_fields=["package_id", "package_name", "city", "days", "price",
                       "total_seats", "remaining_seats", "agency", "rating",
                       "departure_dates", "highlights"],
        # 城市不为空才追加标量过滤；没有城市就传 None，允许在整个集合内检索。
        # 城市值先做 JSON 字符串转义，避免引号改变过滤表达式的结构。
        expr=f"city == {json.dumps(city, ensure_ascii=False)}" if city else None
    )

    # 【检索内部步骤 4】从命中项选取返回字段；虽然检索请求取了更多字段，实际只输出下列字段。
    package_list = []
    # 外层循环遍历每个查询向量对应的命中组，内层遍历该组候选；hit.entity 读取存储字段。
    for hits in results:
        for hit in hits:
            package_list.append({
                "package_id": hit.entity.get("package_id"),
                "package_name": hit.entity.get("package_name"),
                "city": hit.entity.get("city"),
                "days": hit.entity.get("days"),
                "price": hit.entity.get("price"),
                # COSINE 下 hit.distance 表示该检索度量的分数；四舍五入到四位用于展示，不参与二次筛选。
                "similarity": round(hit.distance, 4),
            })
    return package_list


# ==================== 差旅服务类 ====================
class TripService:
    """
    差旅查询与预订服务类

    混合数据源架构：
    - 商务用车、保险：从 MySQL 数据库真实查询
    - 会务套餐：从 Milvus 向量数据库语义检索（RAG）
    """

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

    def _execute_query(self, sql: str, params: list = None) -> str:
        """
        执行 SQL 查询，返回 JSON 格式的结果

        参数：
            sql (str): SQL 查询语句，使用 %s 占位符
            params (list): SQL 参数列表

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

            # 处理特殊类型（日期、Decimal 等）
            # 【主流程 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)

            if results:
                response = {"status": "success", "data": results}
            else:
                response = {"status": "no_data", "message": "未找到差旅数据，请确认查询条件。"}

            # 【主流程 8.4】把 success/no_data 和数据序列化为 JSON 文本交回工具调用方。
            # 异常分支返回 error 文本；工具成功返回文本本身不代表查询有结果。
            return json.dumps(response, cls=DateEncoder, ensure_ascii=False)

        # 查询、取值或序列化出错时进入异常分支，返回含 status="error" 的 JSON 文本。
        # 由于错误被转换为普通返回值，上层还需要检查内容，不能只根据“函数正常返回”判断业务成功。
        except Exception as e:
            logger.error(f"差旅查询错误: {str(e)}")
            return json.dumps({"status": "error", "message": str(e)}, ensure_ascii=False)

    # ========== 商务用车查询（MySQL） ==========

    def query_car_rental(self, pickup_city: str, return_city: str, date: str, car_type: str = None) -> str:
        """
        查询商务用车信息（从 MySQL car_rentals 表查询）
        """
        # 【主流程 8.2】将已传入的参数组织成固定 SQL 和参数列表。
        # 调用者：同名 MCP 包装函数。输入：取车城市、还车城市、取车日期；输出：查询结果的 JSON 字符串。
        # 数据源：MySQL 的 car_rentals 表，SQL 写在代码中，由参数决定筛选值。
        # pickup_city 与 return_city 可以不同；car_type 只有有值时才增加车型筛选。
        logger.info(f"查询商务用车: {pickup_city} -> {return_city}, {date}, {car_type}")
        sql = ("SELECT id, company, pickup_city, return_city, pickup_date, car_type, "
               "car_model, price_per_day, total_available, transmission, seats, deposit "
               "FROM car_rentals "
               "WHERE pickup_city = %s AND return_city = %s AND pickup_date = %s")
        # 参数列表的顺序与 SQL 占位符顺序一致；列表中的日期仍按传入值交给数据库比较。
        params = [pickup_city, return_city, date]
        # 可选条件有值才追加；同时追加参数，保证条件与参数一一对应。
        if car_type:
            sql += " AND car_type = %s"
            # 必须在追加可选 SQL 条件时同时追加对应值，否则占位符数量与参数数量会不一致。
            params.append(car_type)
        # 【主流程 8.3】进入通用查询方法；结果沿当前调用栈原路返回 MCP 包装函数。
        return self._execute_query(sql, params)

    # ========== 会务套餐查询（Milvus RAG） ==========

    def query_meeting_package(self, query_text: str, city: str = None) -> str:
        """
        查询会务套餐信息（从 Milvus 向量数据库进行语义检索）

        参数：
            query_text (str): 用户查询描述，如"成都办数字化转型峰会，需要接送和酒店"、"适合销售团队的短途会务套餐"
            city (str, optional): 城市过滤条件

        返回值：
            str: JSON 字符串，包含匹配的会务套餐列表
        """
        # 【主流程 8.2：会务套餐分支】进入 Milvus 语义检索；用车与保险则走 MySQL。
        # 这里只检索候选套餐，不调用聊天模型生成回答；结果交给 A2A 子助手再整理。
        # 输入 query_text 负责语义相似度，city 负责精确城市过滤，两者用途不同。
        # 固定 limit=5 最多取五个候选；列表非空返回 success，空列表返回 no_data，抛异常返回 error。
        # 本实现没有另外设置相似度阈值，因而返回候选不等于每条都满足用户全部要求。
        logger.info(f"查询会务套餐: query='{query_text}', city='{city}'")
        try:
            package_list = search_meeting_packages_in_milvus(query_text, city, limit=5)

            if package_list:
                return json.dumps({"status": "success", "data": package_list}, ensure_ascii=False)
            else:
                return json.dumps({
                    "status": "no_data",
                    "message": f"未找到匹配的会务套餐。{'请确认城市名称，' if city else ''}或尝试其他搜索条件。"
                }, ensure_ascii=False)

        except Exception as e:
            logger.error(f"会务套餐 RAG 查询错误: {str(e)}")
            return json.dumps({"status": "error", "message": f"会务套餐查询出错：{str(e)}"}, ensure_ascii=False)

    # ========== 保险查询（MySQL） ==========

    def query_insurance(self, insurance_type: str = None) -> str:
        """
        查询差旅保险产品信息（从 MySQL insurances 表查询）
        """
        # 【主流程 8.2】将已传入的参数组织成固定 SQL 和参数列表。
        # 调用者：同名 MCP 包装函数。输入：可选保险类型；输出：查询结果的 JSON 字符串。
        # 数据源：MySQL 的 insurances 表，SQL 写在代码中，由参数决定筛选值。
        # 无类型参数时查询表内所有产品；有类型时追加 WHERE insurance_type = %s。
        logger.info(f"查询保险: {insurance_type}")
        sql = ("SELECT id, insurance_type, name, company, coverage, price, "
               "duration_days, max_coverage, medical_coverage, baggage_coverage, flight_delay "
               "FROM insurances")
        params = []
        # 可选条件有值才追加；同时追加参数，保证条件与参数一一对应。
        if insurance_type:
            sql += " WHERE insurance_type = %s"
            # 必须在追加可选 SQL 条件时同时追加对应值，否则占位符数量与参数数量会不一致。
            params.append(insurance_type)
        # 【主流程 8.3】进入通用查询方法；结果沿当前调用栈原路返回 MCP 包装函数。
        return self._execute_query(sql, params)

    # ========== 预订方法 ==========

    def order_car_rental(self, date: str, car_type: str, number: int) -> str:
        """预订商务用车"""
        # 【主流程 8.2：模拟预订分支】只构造演示文本（部分方法同时记日志）。
        # 这里没有写订单、扣库存、支付、出票或生成保单的动作。
        # 参数用于填充下方 f-string，例如日期、名称、类型和数量会直接显示在演示回复里。
        # 返回字符串后本方法立即结束；它没有再调用查询方法核对库存，也没有生成可持久化的订单对象。
        logger.info(f"预订商务用车: {date}, {car_type}, {number}辆")
        # 【新改动2】明确返回模拟结果，避免被解释为真实下单或库存扣减。
        return f"模拟预订成功，未实际下单，未创建订单，未扣减库存。日期：{date}，车型：{car_type}，数量：{number}辆。"

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

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


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

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

    @trip_mcp.tool(
        name="query_car_rental",
        description="查询商务用车信息，参数：pickup_city(取车城市), return_city(还车城市), date(日期，格式YYYY-MM-DD), car_type(车型类型，可选：经济型/SUV/豪华型/MPV)"
    )
    def query_car_rental(pickup_city: str, return_city: str, date: str, car_type: str = None) -> str:
        # 【主流程 8.1】客户端选中本工具时才进入这里；把命名参数转交业务方法。
        # 调用者：MCP 框架按工具名 query_car_rental 分派到这里。
        # 输入：函数签名中的命名参数；输出：业务方法返回的字符串，再由 MCP 框架封装为工具响应。
        # 同名的 TripService.query_car_rental 是业务方法；当前嵌套函数只是对外暴露的工具入口。
        return service.query_car_rental(pickup_city, return_city, date, car_type)

    @trip_mcp.tool(
        name="query_meeting_package",
        description="查询会务套餐信息（语义搜索），参数：query_text(查询描述，如'成都办数字化转型峰会，需要接送和酒店')，city(城市过滤，可选)"
    )
    def query_meeting_package(query_text: str, city: str = None) -> str:
        # 【主流程 8.1】客户端选中本工具时才进入这里；把命名参数转交业务方法。
        # 调用者：MCP 框架按工具名 query_meeting_package 分派到这里。
        # 输入：函数签名中的命名参数；输出：业务方法返回的字符串，再由 MCP 框架封装为工具响应。
        # 同名的 TripService.query_meeting_package 是业务方法；当前嵌套函数只是对外暴露的工具入口。
        return service.query_meeting_package(query_text, city)

    @trip_mcp.tool(
        name="query_insurance",
        description="查询差旅保险产品，参数：insurance_type(保险类型，可选：综合型/意外型/医疗型/境外型)"
    )
    def query_insurance(insurance_type: str = None) -> str:
        # 【主流程 8.1】客户端选中本工具时才进入这里；把命名参数转交业务方法。
        # 调用者：MCP 框架按工具名 query_insurance 分派到这里。
        # 输入：函数签名中的命名参数；输出：业务方法返回的字符串，再由 MCP 框架封装为工具响应。
        # 同名的 TripService.query_insurance 是业务方法；当前嵌套函数只是对外暴露的工具入口。
        return service.query_insurance(insurance_type)

    # ========== 注册预订工具 ==========

    # 【新改动2】向智能体说明该工具仅用于模拟测试，可以响应模拟预订请求。
    @trip_mcp.tool(
        name="order_car_rental",
        description="根据日期、车型和数量模拟预订商务用车，仅用于测试，不实际下单、不创建订单、不扣减库存。参数：date(商务用车日期，格式YYYY-MM-DD), car_type(车型，如经济型/SUV/豪华型/MPV), number(商务用车数量)"
    )
    def order_car_rental(date: str, car_type: str, number: int) -> str:
        # 【主流程 8.1】客户端选中本工具时才进入这里；把命名参数转交业务方法。
        # 调用者：MCP 框架按工具名 order_car_rental 分派到这里。
        # 输入：函数签名中的命名参数；输出：业务方法返回的字符串，再由 MCP 框架封装为工具响应。
        # 同名的 TripService.order_car_rental 是业务方法；当前嵌套函数只是对外暴露的工具入口。
        logger.info(f"正在预订商务用车: {date}, {car_type}, {number}辆")
        return service.order_car_rental(date, car_type, number)

    @trip_mcp.tool(
        name="order_meeting_package",
        description="仅模拟演示，不实际下单、不创建订单、不扣减库存、不出票或生成保单。根据出发日期、会务套餐名称和人数预订会务套餐。参数：date(出发日期，格式YYYY-MM-DD), package_name(会务套餐名称，需与查询结果中的套餐名称一致), number(预订人数)"
    )
    def order_meeting_package(date: str, package_name: str, number: int) -> str:
        # 【主流程 8.1】客户端选中本工具时才进入这里；把命名参数转交业务方法。
        # 调用者：MCP 框架按工具名 order_meeting_package 分派到这里。
        # 输入：函数签名中的命名参数；输出：业务方法返回的字符串，再由 MCP 框架封装为工具响应。
        # 同名的 TripService.order_meeting_package 是业务方法；当前嵌套函数只是对外暴露的工具入口。
        logger.info(f"正在预订会务套餐: {date}, {package_name}, {number}人")
        return service.order_meeting_package(date, package_name, number)

    @trip_mcp.tool(
        name="order_insurance",
        description="仅模拟演示，不实际下单、不创建订单、不扣减库存、不出票或生成保单。根据保险类型、生效日期和份数购买差旅保险。参数：insurance_type(保险类型，如综合型/意外型/医疗型/境外型), date(生效日期，格式YYYY-MM-DD), number(购买份数)"
    )
    def order_insurance(insurance_type: str, date: str, number: int) -> str:
        # 【主流程 8.1】客户端选中本工具时才进入这里；把命名参数转交业务方法。
        # 调用者：MCP 框架按工具名 order_insurance 分派到这里。
        # 输入：函数签名中的命名参数；输出：业务方法返回的字符串，再由 MCP 框架封装为工具响应。
        # 同名的 TripService.order_insurance 是业务方法；当前嵌套函数只是对外暴露的工具入口。
        logger.info(f"正在购买保险: {insurance_type}, {date}, {number}份")
        return service.order_insurance(insurance_type, date, number)

    # 【启动 S2.3】工具注册结束后构建 HTTP 应用；下一行 uvicorn.run 开始监听并等待请求。
    app = create_fastapi_app(mcp_server=trip_mcp)
    uvicorn.run(app, host="0.0.0.0", port=8003)


if __name__ == '__main__':
    create_trip_mcp_server()