sun
842e8f4f72
- 新增 OpenAIClientModel 类，用于调用 OpenAI 兼容模型 - 重构 AutoToolChatSession 类，支持 OpenAI 兼容模型- 增加了更多日志输出，以便调试和跟踪程序执行流程 - 优化了工具调用和结果处理的逻辑- 调整了环境变量加载方式，使用 dotenv 库
210 lines
10 KiB
Python
210 lines
10 KiB
Python
"""
|
||
FastMCP quickstart example.
|
||
|
||
cd to the `examples/snippets/clients` directory and run:
|
||
uv run server fastmcp_quickstart stdio
|
||
"""
|
||
import os
|
||
from mcp.server.fastmcp import FastMCP
|
||
from dotenv import load_dotenv, find_dotenv
|
||
import requests
|
||
from typing import Dict, Optional, Union
|
||
_ = load_dotenv(find_dotenv())
|
||
# Create an MCP server
|
||
mcp = FastMCP("Demo")
|
||
|
||
|
||
# Add an addition tool
|
||
@mcp.tool()
|
||
def add(a: int, b: int) -> int:
|
||
"""Add two numbers"""
|
||
return a + b
|
||
|
||
@mcp.tool()
|
||
def get_weather_by_location(user_id: str, user_key: str, province: str, place: str) -> Dict[
|
||
str, Union[str, float, int]
|
||
]:
|
||
"""
|
||
调用中国气象局天气预报API，获取指定省份和地点的完整当日天气信息（含基础预报、实时数据、气象预警）
|
||
|
||
### 参数说明
|
||
- user_id: str - 接口调用身份标识，需从http://www.apihz.cn注册获取，不可为空（例："10007673"）
|
||
- user_key: str - 接口通讯秘钥，与user_id对应，注册后获取，不可为空（例："be63d2fb5354f76abb18f62583edffae"）
|
||
- province: str - 查询省份/直辖市，建议去除"省""市"后缀（例："四川"而非"四川省"，"北京"而非"北京市"）
|
||
- place: str - 查询城市/区/县，建议去除"市""区""县"后缀（例："绵阳"而非"绵阳市"，"大兴"而非"大兴区"）
|
||
|
||
### 返回说明（成功，code=200）
|
||
基础预报信息：
|
||
- code: int - 状态码，成功固定为200
|
||
- guo: str - 国家名称（例："中国"）
|
||
- sheng: str - 标准化省份/直辖市名称（例："四川"）
|
||
- shi: str - 标准化城市/地区名称（例："绵阳"）
|
||
- name: str - 与shi一致，冗余字段（例："绵阳"）
|
||
- weather1: str - 当日主要天气（例："阵雨"）
|
||
- weather2: str - 当日次要天气（例："阵雨"）
|
||
- wd1: str - 当日最高温度（单位：℃，例："25"）
|
||
- wd2: str - 当日最低温度（单位：℃，例："18"）
|
||
- winddirection1: str - 当日主要风向（例："无持续风向"）
|
||
- winddirection2: str - 当日次要风向（例："无持续风向"）
|
||
- windleve1: str - 当日主要风力等级（例："微风"）
|
||
- windleve2: str - 当日次要风力等级（例："微风"）
|
||
- weather1img: str - 主要天气图标URL（例："https://rescdn.apihz.cn/resimg/tianqi/zhenyu.png"）
|
||
- weather2img: str - 次要天气图标URL（例："https://rescdn.apihz.cn/resimg/tianqi/zhenyu.png"）
|
||
- lon: str - 地区经度（保留3位小数，例："104.730"）
|
||
- lat: str - 地区纬度（保留3位小数，例："31.440"）
|
||
- uptime: str - 预报数据更新时间（格式：YYYY-MM-DD HH:MM:SS，例："2025-08-29 12:00:00"）
|
||
|
||
实时天气数据（顶层字段）：
|
||
- now_precipitation: float - 当前降水量（单位：mm，例：0.0）
|
||
- now_temperature: float - 当前温度（单位：℃，例：19.3）
|
||
- now_pressure: int - 当前气压（单位：hPa，例：956）
|
||
- now_humidity: int - 当前湿度（单位：%，例：85）
|
||
- now_windDirection: str - 当前风向（例："东北风"）
|
||
- now_windDirectionDegree: int - 当前风向角度（0°=北风，例：28）
|
||
- now_windSpeed: float - 当前风速（单位：m/s，例：3.2）
|
||
- now_windScale: str - 当前风力等级（例："微风"）
|
||
- now_feelst: float - 当前体感温度（单位：℃，例：19.7）
|
||
- now_uptime: str - 实时数据更新时间（格式：YYYY/MM/DD HH:MM，例："2025/08/29 10:05"）
|
||
|
||
气象预警（顶层字段，无预警时为空字符串）：
|
||
- alarm_id: str - 预警唯一ID（例："51070041600000_20250828215515"）
|
||
- alarm_title: str - 预警标题（例："绵阳市气象台更新大风蓝色预警信号[IV级/一般]"）
|
||
- alarm_signaltype: str - 预警类型（例："大风"）
|
||
- alarm_signallevel: str - 预警等级（例："蓝色"）
|
||
- alarm_effective: str - 预警生效时间（例："2025/08/28 21:55"）
|
||
- alarm_eventType: str - 预警事件编码（例："11B06"）
|
||
- alarm_severity: str - 预警等级英文编码（例："BLUE"）
|
||
- alarm_type: str - 预警类型编码（例："p0007004"）
|
||
|
||
### 返回说明（失败，code=400）
|
||
- code: int - 错误状态码，固定为400
|
||
- msg: str - 错误详情（例："通讯秘钥错误。"、"API响应解析失败"）
|
||
"""
|
||
api_url = "https://cn.apihz.cn/api/tianqi/tqyb.php"
|
||
clean_province = province.replace("省", "").replace("市", "").strip()
|
||
clean_place = place.replace("市", "").replace("区", "").replace("县", "").strip()
|
||
final_user_id = os.getenv("APIHZ_ID", user_id)
|
||
final_user_key = os.getenv("APIHZ_KEY", user_key)
|
||
params = {
|
||
"id": final_user_id,
|
||
"key": final_user_key,
|
||
"sheng": clean_province,
|
||
"place": clean_place
|
||
}
|
||
|
||
try:
|
||
# 1. 发送请求（增加超时重试，避免偶发网络波动）
|
||
response = requests.get(api_url, params=params, timeout=10)
|
||
response.raise_for_status() # 捕获4xx/5xx HTTP错误
|
||
|
||
# 2. 解析响应：先判断响应内容是否为空，再转JSON
|
||
response_text = response.text.strip()
|
||
if not response_text:
|
||
return {"code": 400, "msg": "API响应为空，无法解析天气数据"}
|
||
|
||
# 3. 转JSON并防御None（确保weather_data是字典）
|
||
weather_data = response.json()
|
||
if not isinstance(weather_data, dict):
|
||
return {"code": 400, "msg": f"API响应格式错误，不是有效字典（实际类型：{type(weather_data).__name__}）"}
|
||
|
||
# 4. 处理API返回的错误状态（code=400）
|
||
if weather_data.get("code") == 400:
|
||
return {
|
||
"code": 400,
|
||
"msg": weather_data.get("msg", "API返回错误，原因未知")
|
||
}
|
||
# 5. 确保API返回成功状态（code=200）
|
||
elif weather_data.get("code") != 200:
|
||
return {
|
||
"code": 400,
|
||
"msg": f"API返回非成功状态码：{weather_data.get('code', '未知')}"
|
||
}
|
||
|
||
# 6. 提取嵌套数据（确保now_info/alarm_info是字典，避免None）
|
||
now_info = weather_data.get("nowinfo", {})
|
||
if not isinstance(now_info, dict):
|
||
now_info = {} # 若now_info不是字典，强制设为空字典
|
||
|
||
alarm_info = weather_data.get("alarm", {})
|
||
if not isinstance(alarm_info, dict):
|
||
alarm_info = {} # 若alarm不是字典，强制设为空字典
|
||
|
||
# 7. 构造最终返回结果（全部顶层字段，无嵌套）
|
||
return {
|
||
# 基础预报字段
|
||
"code": 200,
|
||
"guo": weather_data.get("guo", ""),
|
||
"sheng": weather_data.get("sheng", ""),
|
||
"shi": weather_data.get("shi", ""),
|
||
"name": weather_data.get("name", ""),
|
||
"weather1": weather_data.get("weather1", ""),
|
||
"weather2": weather_data.get("weather2", ""),
|
||
"wd1": weather_data.get("wd1", ""),
|
||
"wd2": weather_data.get("wd2", ""),
|
||
"winddirection1": weather_data.get("winddirection1", ""),
|
||
"winddirection2": weather_data.get("winddirection2", ""),
|
||
"windleve1": weather_data.get("windleve1", ""),
|
||
"windleve2": weather_data.get("windleve2", ""),
|
||
"weather1img": weather_data.get("weather1img", ""),
|
||
"weather2img": weather_data.get("weather2img", ""),
|
||
"lon": weather_data.get("lon", ""),
|
||
"lat": weather_data.get("lat", ""),
|
||
"uptime": weather_data.get("uptime", ""),
|
||
# 实时天气字段
|
||
"now_precipitation": float(now_info.get("precipitation", 0.0)),
|
||
"now_temperature": float(now_info.get("temperature", 0.0)),
|
||
"now_pressure": int(now_info.get("pressure", 0)),
|
||
"now_humidity": int(now_info.get("humidity", 0)),
|
||
"now_windDirection": now_info.get("windDirection", ""),
|
||
"now_windDirectionDegree": int(now_info.get("windDirectionDegree", 0)),
|
||
"now_windSpeed": float(now_info.get("windSpeed", 0.0)),
|
||
"now_windScale": now_info.get("windScale", ""),
|
||
"now_feelst": float(now_info.get("feelst", 0.0)),
|
||
"now_uptime": now_info.get("uptime", ""),
|
||
# 预警字段
|
||
"alarm_id": alarm_info.get("id", ""),
|
||
"alarm_title": alarm_info.get("title", ""),
|
||
"alarm_signaltype": alarm_info.get("signaltype", ""),
|
||
"alarm_signallevel": alarm_info.get("signallevel", ""),
|
||
"alarm_effective": alarm_info.get("effective", ""),
|
||
"alarm_eventType": alarm_info.get("eventType", ""),
|
||
"alarm_severity": alarm_info.get("severity", ""),
|
||
"alarm_type": alarm_info.get("type", "")
|
||
}
|
||
|
||
# 8. 捕获各类异常（明确错误原因）
|
||
except requests.exceptions.HTTPError as e:
|
||
status_code = response.status_code if "response" in locals() else "未知"
|
||
return {"code": 400, "msg": f"HTTP请求错误（状态码：{status_code}）：{str(e)}"}
|
||
except requests.exceptions.ConnectionError:
|
||
return {"code": 400, "msg": "网络连接错误：无法连接到天气API服务器"}
|
||
except requests.exceptions.Timeout:
|
||
return {"code": 400, "msg": "请求超时：API服务器10秒内未响应"}
|
||
except ValueError as e:
|
||
# 捕获JSON解析错误（如响应不是合法JSON）
|
||
return {"code": 400, "msg": f"API响应解析失败（JSON格式错误）：{str(e)}"}
|
||
except Exception as e:
|
||
# 捕获其他未知错误（附带具体错误信息，便于调试）
|
||
return {"code": 400, "msg": f"未知错误：{str(e)}（错误类型：{type(e).__name__}）"}
|
||
|
||
# Add a dynamic greeting resource
|
||
@mcp.resource("greeting://{name}")
|
||
def get_greeting(name: str) -> str:
|
||
"""Get a personalized greeting"""
|
||
return f"Hello, {name}!"
|
||
|
||
|
||
# Add a prompt
|
||
@mcp.prompt()
|
||
def greet_user(name: str, style: str = "friendly") -> str:
|
||
"""Generate a greeting prompt"""
|
||
styles = {
|
||
"friendly": "Please write a warm, friendly greeting",
|
||
"formal": "Please write a formal, professional greeting",
|
||
"casual": "Please write a casual, relaxed greeting",
|
||
}
|
||
|
||
return f"{styles.get(style, styles['friendly'])} for someone named {name}."
|
||
|
||
if __name__ == "__main__":
|
||
mcp.run(transport="sse") |