"""
MCP Weather Server
一個基於 OpenWeatherMap API 的天氣 MCP 伺服器
提供當前天氣資訊和天氣預報的功能
https://www.claudemcp.com/zh/docs/mcp-py-sdk-basic
未知錯誤: 'locale' codec can't encode character '\u6708' in position 2: encoding error
https://blog.csdn.net/lly1122334/article/details/116200252
"""
import os
from typing import Dict, Any, List
from datetime import datetime
import requests
from dotenv import load_dotenv
from mcp.server.fastmcp import FastMCP
import sys
import io
import locale
sys.stdout = io.TextIOWrapper(sys.stdout.buffer, encoding='utf-8')
locale.setlocale(locale.LC_CTYPE, 'Chinese')
# 載入環境變數
load_dotenv()
# 建立 MCP 伺服器
mcp = FastMCP("Weather")
# OpenWeatherMap API 配置
OPENWEATHER_API_KEY = os.getenv("OPENWEATHER_API_KEY")
if not OPENWEATHER_API_KEY:
print("警告: 找不到 OPENWEATHER_API_KEY 環境變數")
print("請在 .env 檔案中設定你的 OpenWeatherMap API 金鑰")
OPENWEATHER_BASE_URL = "https://api.openweathermap.org/data/2.5"
def format_temperature(temp_kelvin: float) -> str:
"""格式化溫度顯示（開爾文轉攝氏)"""
celsius = temp_kelvin - 273.15
fahrenheit = celsius * 9/5 + 32
return f"{celsius:.1f}°C ({fahrenheit:.1f}°F)"
def format_weather_info(weather_data: Dict[str, Any]) -> str:
"""格式化天氣資訊為易讀的字串"""
main = weather_data.get("main", {})
weather = weather_data.get("weather", [{}])[0]
wind = weather_data.get("wind", {})
clouds = weather_data.get("clouds", {})
location = weather_data.get("name", "未知位置")
country = weather_data.get("sys", {}).get("country", "")
if country:
location += f", {country}"
# 基本天氣資訊
description = weather.get("description", "").title()
temp = format_temperature(main.get("temp", 0))
feels_like = format_temperature(main.get("feels_like", 0))
# 詳細資訊
humidity = main.get("humidity", 0)
pressure = main.get("pressure", 0)
wind_speed = wind.get("speed", 0)
wind_deg = wind.get("deg", 0)
cloudiness = clouds.get("all", 0)
# 可見度（以公尺為單位，轉換為公里）
visibility = weather_data.get("visibility", 0) / 1000
result = f"""🌍 **{location}**
🌤️ **當前天氣**: {description}
🌡️ **溫度**: {temp}
🤒 **體感溫度**: {feels_like}
💧 **濕度**: {humidity}%
🌪️ **氣壓**: {pressure} hPa
💨 **風速**: {wind_speed} m/s
🧭 **風向**: {wind_deg}°
☁️ **雲量**: {cloudiness}%
👁️ **能見度**: {visibility:.1f} km"""
# 加入日出日落時間（如果有的話）
sys_info = weather_data.get("sys", {})
if "sunrise" in sys_info and "sunset" in sys_info:
sunrise = datetime.fromtimestamp(sys_info["sunrise"]).strftime("%H:%M")
sunset = datetime.fromtimestamp(sys_info["sunset"]).strftime("%H:%M")
result += f"\n🌅 **日出**: {sunrise}"
result += f"\n🌇 **日落**: {sunset}"
return result
def format_forecast_info(forecast_data: Dict[str, Any]) -> str:
"""格式化天氣預報訊息"""
city = forecast_data.get("city", {})
location = city.get("name", "未知位置")
country = city.get("country", "")
if country:
location += f", {country}"
forecasts = forecast_data.get("list", [])
result = f"📅 **{location} - 5天天氣預報**\n\n"
# 按日期分組預報數據
daily_forecasts: Dict[str, List[Dict[str, Any]]] = {}
for forecast in forecasts:
dt = datetime.fromtimestamp(forecast["dt"])
date_key = dt.strftime("%Y-%m-%d")
if date_key not in daily_forecasts:
daily_forecasts[date_key] = []
daily_forecasts[date_key].append(forecast)
# 顯示每天的天氣預報
for date_key, day_forecasts in list(daily_forecasts.items())[:5]: # 只顯示5天
date_obj = datetime.strptime(date_key, "%Y-%m-%d")
date_str = date_obj.strftime("%m月%d日 (%A)")
result += f"**{date_str}**\n"
# 取得當天的溫度範圍
temps = [f["main"]["temp"] for f in day_forecasts]
min_temp = format_temperature(min(temps))
max_temp = format_temperature(max(temps))
# 取得主要天氣描述（出現頻率最高的）
descriptions = [f["weather"][0]["description"] for f in day_forecasts]
main_desc = max(set(descriptions), key=descriptions.count).title()
# 取得平均濕度和風速
avg_humidity = sum(f["main"]["humidity"] for f in day_forecasts) / len(day_forecasts)
avg_wind_speed = sum(f["wind"]["speed"] for f in day_forecasts) / len(day_forecasts)
result += f" 🌤️ {main_desc}\n"
result += f" 🌡️ {min_temp} - {max_temp}\n"
result += f" 💧 濕度: {avg_humidity:.0f}%\n"
result += f" 💨 風速: {avg_wind_speed:.1f} m/s\n\n"
return result
@mcp.tool()
def get_current_weather(city: str) -> str:
"""
獲取指定城市的當前天氣信息
Args:
city: 城市名稱（英文或中文）
Returns:
格式化的當前天氣資訊
"""
if not OPENWEATHER_API_KEY:
return "❌ 錯誤: 未配置 OpenWeatherMap API 金鑰。請設定 OPENWEATHER_API_KEY 環境變數。"
print(f"正在取得 {city} 的當前天氣資訊...")
try:
response = requests.get(
f"{OPENWEATHER_BASE_URL}/weather",
params={
"q": city,
"appid": OPENWEATHER_API_KEY,
"lang": "zh_tw"
},
timeout=10
)
if response.status_code == 404:
return f"❌ 錯誤: 找不到城市 '{city}'。請檢查城市名稱是否正確。"
elif response.status_code == 401:
return "❌ 錯誤: API 金鑰無效。請檢查 OPENWEATHER_API_KEY 配置。"
elif response.status_code != 200:
return f"❌ 錯誤: API 請求失敗 (狀態碼: {response.status_code})"
weather_data = response.json()
return format_weather_info(weather_data)
except requests.RequestException as e:
return f"❌ 網路錯誤: {str(e)}"
except Exception as e:
return f"❌ 未知錯誤: {str(e)}"
@mcp.tool()
def get_weather_forecast(city: str, days: int = 5) -> str:
"""
取得指定城市的天氣預報
Args:
city: 城市名稱（英文或中文）
days: 預報天數（1-5天，預設5天）
Returns:
格式化的天氣預報訊息
"""
if not OPENWEATHER_API_KEY:
return "❌ 錯誤: 未配置 OpenWeatherMap API 金鑰。請設定 OPENWEATHER_API_KEY 環境變數。"
if days < 1 or days > 5:
return "❌ 錯誤: 預報天數必須在 1-5 天之間。"
print(f"正在取得 {city} 的 {days} 天天氣預報...")
try:
response = requests.get(
f"{OPENWEATHER_BASE_URL}/forecast",
params={
"q": city,
"appid": OPENWEATHER_API_KEY,
"lang": "zh_tw"
},
timeout=10
)
if response.status_code == 404:
return f"❌ 錯誤: 找不到城市 '{city}'。請檢查城市名稱是否正確。"
elif response.status_code == 401:
return "❌ 錯誤: API 金鑰無效。請檢查 OPENWEATHER_API_KEY 配置。"
elif response.status_code != 200:
return f"❌ 錯誤: API 請求失敗 (狀態碼: {response.status_code})"
forecast_data = response.json()
return format_forecast_info(forecast_data)
except requests.RequestException as e:
return f"❌ 網路錯誤: {str(e)}"
except Exception as e:
return f"❌ 未知錯誤: {str(e)}"
@mcp.resource("weather://current/{city}")
def get_current_weather_resource(city: str) -> str:
"""取得指定城市目前天氣的資源"""
return f"目前天氣資訊資源: {city}"
@mcp.resource("weather://forecast/{city}")
def get_forecast_resource(city: str) -> str:
"""取得指定城市天氣預報的資源"""
return f"天氣預報資源: {city}"
@mcp.resource("weather://api-status")
def get_api_status() -> str:
"""取得 API 狀態資訊"""
if OPENWEATHER_API_KEY:
return "✅ OpenWeatherMap API 金鑰已配置"
else:
return "❌ OpenWeatherMap API 金鑰未配置"
def main():
"""運行 MCP 伺服器"""
print("🌤️ 啟動天氣 MCP 伺服器...")
print("📍 支援的功能:")
print(" - 取得目前天氣 (get_current_weather)")
print(" - 取得天氣預報 (get_weather_forecast)")
print()
if not OPENWEATHER_API_KEY:
print("⚠️ 警告: 未配置 OpenWeatherMap API 金鑰")
print("請建立 .env 檔案並新增以下內容:")
print("OPENWEATHER_API_KEY=your_api_key_here")
print()
print("取得 API 金鑰: https://openweathermap.org/api")
print()
mcp.run(transport='stdio')
if __name__ == "__main__":
main()
