freebird2913

1
def get_location_id(city_name: str) -> str:
2
    """
3
    根据城市名称获取和风天气的LocationID
4
    """
5
    url = "https://geoapi.qweather.com/v2/city/lookup"
6
    params = {
7
        "location": city_name,
8
        "key": os.getenv("QWEATHER_API_KEY")
9
    }
10

11
    try:
12
        response = requests.get(url, params=params, timeout=10)
13
        data = response.json()
14

15
        if data.get("code") == "200" and data.get("location"):
16
            return data["location"][0]["id"]
17
        else:
18
            return None
19
    except Exception as e:
20
        print(f"查询城市ID失败: {e}")
21
        return None
22

23

24
def fetch_weather(location_id: str) -> dict:
25
    """
26
    根据LocationID获取实时天气数据
27
    """
28
    url = "https://devapi.qweather.com/v7/weather/now"
29
    params = {
30
        "location": location_id,
31
        "key": os.getenv("QWEATHER_API_KEY")
32
    }
33

34
    try:
35
        response = requests.get(url, params=params, timeout=10)
36
        data = response.json()
37

38
        if data.get("code") == "200":
39
            return data.get("now", {})
40
        else:
41
            return {"error": f"API返回错误码: {data.get('code')}"}
42
    except Exception as e:
43
        return {"error": f"请求失败: {e}"}

1
# server.py 完整代码
2
from mcp.server.fastmcp import FastMCP
3
from dotenv import load_dotenv
4
import os
5
import requests
6
import json
7

8
load_dotenv()
9

10
# 创建MCP服务器实例
11
mcp = FastMCP("WeatherServer")
12

13

14
def get_location_id(city_name: str) -> str:
15
    """根据城市名称获取和风天气的LocationID"""
16
    url = "https://geoapi.qweather.com/v2/city/lookup"
17
    params = {
18
        "location": city_name,
19
        "key": os.getenv("QWEATHER_API_KEY")
20
    }
21

22
    try:
23
        response = requests.get(url, params=params, timeout=10)
24
        data = response.json()
25

26
        if data.get("code") == "200" and data.get("location"):
27
            return data["location"][0]["id"]
28
        return None
29
    except Exception as e:
30
        print(f"查询城市ID失败: {e}")
31
        return None
32

33

34
def fetch_weather(location_id: str) -> dict:
35
    """根据LocationID获取实时天气数据"""
36
    url = "https://devapi.qweather.com/v7/weather/now"
37
    params = {
38
        "location": location_id,
39
        "key": os.getenv("QWEATHER_API_KEY")
40
    }
41

42
    try:
43
        response = requests.get(url, params=params, timeout=10)
44
        data = response.json()
45

46
        if data.get("code") == "200":
47
            return data.get("now", {})
48
        return {"error": f"API返回错误码: {data.get('code')}"}
49
    except Exception as e:
50
        return {"error": f"请求失败: {e}"}
51

52

53
@mcp.tool()
54
def get_current_weather(city_name: str) -> str:
55
    """
56
    获取指定城市的当前天气信息
57

58
    Args:
59
        city_name: 城市名称，例如"北京"、"上海"、"广州"
60

61
    Returns:
62
        包含温度、湿度、天气状况等信息的字符串
63
    """
64
    # 获取LocationID
65
    location_id = get_location_id(city_name)
66
    if not location_id:
67
        return f"抱歉，没有找到城市「{city_name}」的天气信息，请检查城市名称是否正确。"
68

69
    # 获取天气数据
70
    weather_data = fetch_weather(location_id)
71

72
    if "error" in weather_data:
73
        return weather_data["error"]
74

75
    # 格式化返回结果
76
    result = f"""
77
📍 城市：{city_name}
78
🌡️ 温度：{weather_data.get('temp', 'N/A')}°C
79
💧 湿度：{weather_data.get('humidity', 'N/A')}%
80
☁️ 天气：{weather_data.get('text', 'N/A')}
81
💨 风向：{weather_data.get('windDir', 'N/A')}
82
🌀 风力等级：{weather_data.get('windScale', 'N/A')}级
83
👁️ 能见度：{weather_data.get('vis', 'N/A')} km
84
    """
85
    return result.strip()
86

87

88
@mcp.tool()
89
def get_weather_by_location(latitude: float, longitude: float) -> str:
90
    """
91
    根据经纬度获取当前天气信息
92

93
    Args:
94
        latitude: 纬度，例如 39.9042
95
        longitude: 经度，例如 116.4074
96

97
    Returns:
98
        包含温度、湿度、天气状况等信息的字符串
99
    """
100
    # 构建经纬度格式
101
    location = f"{longitude},{latitude}"
102

103
    url = "https://devapi.qweather.com/v7/weather/now"
104
    params = {
105
        "location": location,
106
        "key": os.getenv("QWEATHER_API_KEY")
107
    }
108

109
    try:
110
        response = requests.get(url, params=params, timeout=10)
111
        data = response.json()
112

113
        if data.get("code") != "200":
114
            return f"获取天气信息失败，错误码: {data.get('code')}"
115

116
        now = data.get("now", {})
117
        result = f"""
118
📍 位置：{latitude}, {longitude}
119
🌡️ 温度：{now.get('temp', 'N/A')}°C
120
💧 湿度：{now.get('humidity', 'N/A')}%
121
☁️ 天气：{now.get('text', 'N/A')}
122
💨 风向：{now.get('windDir', 'N/A')}
123
🌀 风力等级：{now.get('windScale', 'N/A')}级
124
👁️ 能见度：{now.get('vis', 'N/A')} km
125
        """
126
        return result.strip()
127
    except Exception as e:
128
        return f"请求失败: {e}"
129

130

131
if __name__ == "__main__":
132
    # 使用STDIO协议运行服务器（本地调试用）
133
    mcp.run(transport='stdio')