SmartBizTrip/mcp_server/mcp_weather_server.py

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

# ============================================================
# MCP 天气服务模块
# 功能：提供天气查询的 MCP（Model Context Protocol）工具接口
# 支持两种数据源：MySQL 数据库 / 和风天气 API（通过配置切换）
# 启动入口：文件末尾 __main__；运行环境需要能导入 SmartBizTrip 包
# 访问地址：http://127.0.0.1:8002/mcp
#
# 调用流程图：
#
#   AI 智能体（如 Claude）
#        │
#        ▼
#   ┌─────────────────────────────┐
#   │  MCP Server (端口 8002)      │
#   │  create_weather_mcp_server() │
#   │  FastMCP("WeatherTools")     │
#   └──────────────┬──────────────┘
#                  │
#                  ▼
#           query_weather(city, start_date, end_date)
#                  │
#        ┌─────────┴─────────┐
#        │ 检查 weather_source│
#        └─────────┬─────────┘
#                  │
#        ┌─────────┴─────────┐
#        │                   │
#     "api"               其他（"db"）
#        │                   │
#        ▼                   ▼
#   ┌──────────────┐  ┌──────────────────┐
#   │ fetch_weather│  │ WeatherService   │
#   │ _from_api()  │  │ .query_weather() │
#   └──────┬───────┘  └────────┬─────────┘
#          │                   │
#          ▼                   ▼
#   ① geo接口:城市→ID     SQL查询 weather_data 表
#          │           SELECT ... FROM weather_data
#          ▼            WHERE city = %s AND
#   ② 天气API:用ID获取数据    fx_date BETWEEN %s AND %s
#          │                   │
#          ▼                   ▼
#   ③ 过滤日期范围         遍历结果集（字典格式）
#   ④ 字段名映射           处理 date/Decimal 类型
#      tempMax→temp_max      │
#      textDay→text_day      ▼
#                     json.dumps() 序列化为 JSON 字符串
#          │                   │
#          └─────────┬─────────┘
#                    ▼
#           JSON 字符串返回给 AI 智能体
#
#   返回格式示例：
#   {
#     "status": "success",
#     "data": [
#       {
#         "city": "成都",
#         "fx_date": "2026-04-29",
#         "temp_max": 28,
#         "temp_min": 18,
#         "text_day": "多云",
#         "text_night": "晴",
#         "humidity": 65,
#         "wind_dir_day": "南风",
#         "wind_scale_day": "3",
#         "precip": 0.0
#       }
#     ]
#   }
#
# 数据源切换：修改 config.py 中的 weather_source 配置
#   weather_source = "api"  → 使用和风天气实时 API
#   weather_source = "db"   → 使用本地 MySQL 数据库
# ============================================================

import mysql.connector  # MySQL 数据库驱动，用于连接和查询本地天气数据
import json             # JSON 序列化/反序列化，将 Python 字典转为 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                       # 项目配置类，读取数据库/API 等配置信息
from create_logger import logger                # 日志记录器，统一输出运行日志
from utils.format import DateEncoder, default_encoder  # 自定义 JSON 编码器，处理 date/Decimal 等特殊类型
import requests  # HTTP 请求库，用于调用和风天气 API

# 加载全局配置对象
conf = Config()


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

    def query_weather(self, city: str, start_date: str, end_date: str) -> str:
        """
        参数化查询数据库天气数据，返回 JSON 字符串

        参数：
            city: 城市名称，如 "成都"
            start_date: 开始日期，格式 "YYYY-MM-DD"
            end_date: 结束日期，格式 "YYYY-MM-DD"

        返回：
            JSON 字符串，包含 status（状态）和 data（天气数据列表）
        """
        # 【主流程 8.3，数据库内部步骤 1】创建字典游标，SQL 与参数分别传给 execute。
        # %s 是数据库驱动的参数占位符；参数顺序要与 SQL 中占位符顺序一致。
        # 调用者：MCP 包装函数在数据库分支时调用。city 是城市名，日期区间由 start_date/end_date 指定。
        # start_date=end_date 表示查某一天；BETWEEN 包含开始和结束日期，ORDER BY fx_date 按日期升序返回。
        # SELECT 明确选择天气字段，所以结果不包含 weather_data 表的全部列。
        try:
            # 创建游标，dictionary=True 使返回结果为字典而非元组
            # 游标是这一次数据库操作的执行入口；dictionary=True 让每行结果按列名取值。
            # 例：{"city": "北京", "fx_date": 日期对象}，比按 results[0][1] 这样的列位置读取更直观。
            cursor = self.conn.cursor(dictionary=True)

            # 参数化 SQL 查询：使用 %s 占位符防止 SQL 注入攻击
            # 查询 weather_data 表中指定城市和时间范围内的天气记录
            sql = ("SELECT city, fx_date, temp_max, temp_min, text_day, text_night, "
                   "humidity, wind_dir_day, wind_scale_day, precip FROM weather_data "
                   "WHERE city = %s AND fx_date BETWEEN %s AND %s ORDER BY fx_date")
            # 执行查询，参数以列表形式传入，自动转义防止注入
            # SQL 字符串描述查哪些表和条件，参数列表只提供值，二者由驱动组合。
            # 不要把这里的 %s 当成 Python 字符串格式化去替换；当前代码没有将用户输入拼进 SQL 语法。
            cursor.execute(sql, [city, start_date, end_date])

            # 获取所有匹配的查询结果
            # 【主流程 8.3，数据库内部步骤 2】取回全部匹配行，再关闭此次查询的游标。
            results = cursor.fetchall()
            # 关闭当前游标，self.conn 仍保留供下一次工具调用使用；这里没有关闭整条数据库连接。
            cursor.close()  # 关闭游标释放资源

            # 处理特殊类型字段：date/Decimal 等类型无法直接 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)  # 日期/时间转为字符串，Decimal 转为 float

            # 根据是否有结果返回不同格式：
            # - 有数据：status="success" + data 列表
            # - 无数据：status="no_data" + 提示信息
            # ensure_ascii=False 保证中文正常显示而非 Unicode 转义
            # 【主流程 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
            )
        # 查询、取值或序列化出错时进入异常分支，返回含 status="error" 的 JSON 文本。
        # 由于错误被转换为普通返回值，上层还需要检查内容，不能只根据“函数正常返回”判断业务成功。
        except Exception as e:
            # 捕获异常，记录日志并返回错误信息
            logger.error(f"天气查询错误: {str(e)}")
            return json.dumps({"status": "error", "message": str(e)}, ensure_ascii=False)


def fetch_weather_from_api(city: str, start_date: str, end_date: str) -> str:
    """
    从和风天气 API 获取天气数据

    工作流程：
        1. 先用城市名调用 geo 接口获取 location_id
        2. 再用 location_id 调用天气预报接口获取数据
        3. 过滤出指定日期范围内的数据，格式化后返回

    参数：
        city: 城市名称，如 "成都"
        start_date: 开始日期，格式 "YYYY-MM-DD"
        end_date: 结束日期，格式 "YYYY-MM-DD"

    返回：
        JSON 字符串，格式与数据库查询一致
    """
    # 【天气 API 内部步骤 1】用城市名查 location_id；找不到城市时立即返回 no_data。
    # 调用者：MCP 包装函数在 api 分支时调用。输入仍是城市和日期区间。
    # 输出与数据库分支一样采用 JSON 文本，让上游可以按同一种 status/data/message 结构读取。
    # 两次 requests.get 按先后执行：没有城市 ID，就不会请求天气预报。
    logger.info(f"从和风天气API获取数据: {city}, {start_date} ~ {end_date}")

    # 设置请求头，传入和风天气 API Key（认证凭证）
    # X-QW-Api-Key 是请求认证头；密钥从配置读取，城市与日期不是放在这个头里。
    headers = {
        "X-QW-Api-Key": conf.weather_api_key
    }

    # 第一步：城市名 → 地理位置 ID（和风天气需要先查城市 ID）
    geo_url = f"https://{conf.weather_api_host}/geo/v2/city/lookup?location={city}"
    geo_resp = requests.get(geo_url, headers=headers, timeout=15)
    geo_resp.raise_for_status()
    # 把 HTTP 响应体解析为 Python 字典；HTTP 响应对象与解析后的字典是两个不同层次。
    geo_data = geo_resp.json()

    # 如果 API 没有找到该城市，直接返回错误信息
    # 没有 location 字段或列表为空都提前返回 no_data，避免下一行取第一个元素时越界。
    if not geo_data.get("location"):
        return json.dumps({"status": "no_data", "message": f"未找到城市：{city}"}, ensure_ascii=False)

    # 取第一个匹配的城市 ID
    # 若城市查找返回多个候选，本实现直接采用第一个候选，并不再询问用户选择。
    location_id = geo_data["location"][0]["id"]

    # 第二步：用 location_id 获取天气预报数据
    # 【天气 API 内部步骤 2】用 location_id 请求预报；可查范围受配置 URL 的预报天数限制。
    weather_url = f"{conf.weather_base_url}?location={location_id}"
    weather_resp = requests.get(weather_url, headers=headers, timeout=15)
    weather_resp.raise_for_status()
    weather_data = weather_resp.json()

    # 第三步：过滤日期范围，并将 API 返回的字段映射为数据库同名字段
    results = []
    # 【天气 API 内部步骤 3】仅保留请求区间内的返回日期，统一字段名后交回 MCP。
    # 有返回数据也不保证覆盖整个请求区间；代码未在这里检查日期是否齐全。
    for day in weather_data.get("daily", []):
        # 字符串比较即可，因为日期格式都是 "YYYY-MM-DD"
        # 只有规范的 YYYY-MM-DD 格式才适合这种字符串日期比较；当前函数没有额外进行日期格式校验。
        # 不在区间内的 day 不进入 results，继续处理下一天。
        if start_date <= day.get("fxDate", "") <= end_date:
            # 将 API 的驼峰字段统一成数据库风格字段，并把温度/湿度/降水量转成数值。
            # 部分字段缺失时 get 默认 0 或空串，这是代码的兜底值，不能理解成接口确认的天气数值。
            results.append({
                "city": city,                              # 城市名
                "fx_date": day.get("fxDate"),              # 预报日期
                "temp_max": int(day.get("tempMax", 0)),    # 最高温度（转为整数）
                "temp_min": int(day.get("tempMin", 0)),    # 最低温度（转为整数）
                "text_day": day.get("textDay", ""),        # 白天天气描述
                "text_night": day.get("textNight", ""),    # 夜间天气描述
                "humidity": int(day.get("humidity", 0)),   # 湿度百分比
                "wind_dir_day": day.get("windDirDay", ""), # 白天风向
                "wind_scale_day": day.get("windScaleDay", ""),  # 白天风力等级
                "precip": float(day.get("precip", 0))      # 降水量（保留浮点）
            })

    # 根据是否有结果返回对应格式
    if results:
        return json.dumps({"status": "success", "data": results}, ensure_ascii=False)
    return json.dumps({"status": "no_data", "message": "未找到天气数据。"}, ensure_ascii=False)

def create_weather_mcp_server():
    """
    创建并启动天气 MCP 服务器

    MCP（Model Context Protocol）：一种协议，允许 AI 模型（如 Claude）通过标准接口调用外部工具
    本模块将天气查询功能封装为 MCP 工具，供 AI 智能体调用
    """
    # 创建 FastMCP 实例，相当于定义一个工具服务
    # TODO 1.构建FastMCP实例
    # 【启动 S2.1】创建工具注册表。下面的 @tool 在定义包装函数时完成注册。
    # 注册时不会运行包装函数体；只有收到该工具的调用请求才执行函数体。
    # 这是服务启动函数。内部定义的工具包装函数会记住当前的 service 对象，后续每次调用可继续使用它。
    # 这叫闭包：包装函数虽然在外层函数内部定义，却能读取外层的 service。
    # FastMCP 注册这些函数后，HTTP 请求按工具名选择其中一个，不按源代码排列顺序逐个执行。
    weather_mcp = FastMCP(
        name="WeatherTools",  # 工具服务名称，AI 调用时会看到这个名称
    )

    # 实例化数据库查询服务
    # 【启动 S2.2】创建可被所有工具包装函数引用的业务服务，并建立 MySQL 连接。
    # API 模式不创建数据库连接；只有数据库模式才需要 WeatherService。
    service = WeatherService() if conf.weather_source != "api" else None

    # TODO 2.基于MCP实例对象，使用装饰器的方式注册可用工具
    # TODO description非常重要，尤其是日期的格式，需要明确指定。不然调用API可能会失败
    # 使用 @tool 装饰器将函数注册为 MCP 工具
    # name: 工具的唯一标识，AI 通过这个名字调用
    # description: 工具说明，AI 会阅读这段描述来判断是否使用该工具
    @weather_mcp.tool(
        name="query_weather",
        description="查询天气数据，参数：city(城市), start_date(开始日期YYYY-MM-DD), end_date(结束日期YYYY-MM-DD)"
    )
    def query_weather(city: str, start_date: str, end_date: str) -> str:
        """
        MCP 工具入口函数：根据配置选择从数据库或 API 获取天气数据

        参数：
            city: 城市名称，如 "成都"
            start_date: 开始日期，格式 "YYYY-MM-DD"
            end_date: 结束日期，格式 "YYYY-MM-DD"

        返回：
            JSON 字符串，包含天气数据
        """
        # 【主流程 8.1】客户端选中本工具时才进入这里；把命名参数转交业务方法。
        # 调用者：MCP 框架按工具名 query_weather 分派到这里。
        # 输入：函数签名中的命名参数；输出：业务方法返回的字符串，再由 MCP 框架封装为工具响应。
        # 同名的 WeatherService.query_weather 是业务方法；当前嵌套函数只是对外暴露的工具入口。
        logger.info(f"执行天气查询: {city}, {start_date} ~ {end_date}")

        # 根据 config.py 中的 weather_source 配置决定数据源：
        # - "api": 调用和风天气实时 API
        # - 其他值（如 "db"）: 查询本地 MySQL 数据库
        # 【主流程 8.2】这是互斥分支：api 走远程预报接口，其余值走 MySQL。
        if conf.weather_source == "api":
            return fetch_weather_from_api(city, start_date, end_date)
        return service.query_weather(city, start_date, end_date)

    # 打印服务启动信息到日志
    logger.info("=== 天气MCP服务器信息 ===")
    logger.info(f"名称: {weather_mcp.name}")

    # TODO 3.启动fastmcp server : 把MCP Server对象转成fastapi服务， 通过uvicorn启动fastapi
    try:
        # 提示用户访问地址
        print("服务器已启动，请访问 http://127.0.0.1:8002/mcp")

        # 将 FastMCP 对象转为 FastAPI 应用
        # 【启动 S2.3】工具注册结束后构建 HTTP 应用；下一行 uvicorn.run 开始监听并等待请求。
        weather_mcp_server = create_fastapi_app(weather_mcp)

        # 使用 uvicorn 启动 HTTP 服务
        # host="0.0.0.0" 表示允许外部访问，port=8002 是服务端口
        uvicorn.run(weather_mcp_server, host="0.0.0.0", port=8002)
    except Exception as e:
        # 捕获启动异常
        print(f"服务器启动失败: {e}")

if __name__ == '__main__':
    # 调试用法：直接调用 API 测试，不启动 MCP 服务
    # api = fetch_weather_from_api(city="成都", start_date="2026-04-28", end_date="2026-05-01")
    # print(api)

    # 启动 MCP 天气服务器
    create_weather_mcp_server()