Weather MCP

from datetime import datetime from typing import Any, Literal from pydantic import BaseModel, Field class WeatherCurrentInput(BaseModel): # `location` comes from QWeather spec and supports: # - LocationID (e.g. 101010100) # - longitude,latitude (e.g. 116.41,39.92) # - city name (e.g. 北京 / Beijing) # We keep it as a single string to preserve flexibility and avoid premature coupling. location: str = Field(min_length=1, max_length=128) # QWeather supports multi-language and unit conversions. # Defaults align with common usage in China. lang: str = "zh" unit: Literal["m", "i"] = "m" # m=metric, i=imperial class WeatherCurrentResult(BaseModel): # Unified MCP output structure location: str observed_at: datetime temperature: float condition: str humidity: int wind_speed: float wind_dir: str # Keep raw response for debugging / traceability. raw: dict[str, Any] def map_qweather_now_to_result(*, location: str, raw: dict[str, Any]) -> WeatherCurrentResult: """Map QWeather `/v7/weather/now` payload into MCP output. Why map this way: - Upstream JSON has many fields; MCP only exposes stable, commonly-used fields. - `raw` preserves the full payload for debugging and future extensions without breaking the API contract. Field mapping (QWeather -> MCP): - now.obsTime -> observed_at - now.temp -> temperature - now.text -> condition - now.humidity -> humidity - now.windSpeed -> wind_speed - now.windDir -> wind_dir """ now: dict[str, Any] = raw.get("now") or {} obs_time = datetime.fromisoformat(str(now.get("obsTime"))) return WeatherCurrentResult( location=location, observed_at=obs_time, temperature=float(now.get("temp")), condition=str(now.get("text")), humidity=int(now.get("humidity")), wind_speed=float(now.get("windSpeed")), wind_dir=str(now.get("windDir")), raw=raw, ) class WeatherForecastDailyInput(BaseModel): location: str = Field(min_length=1, max_length=128) days: Literal[3, 7, 10, 15] = 3 lang: str = "zh" unit: Literal["m", "i"] = "m" class WeatherDailyForecastItem(BaseModel): date: str temp_max: float temp_min: float condition_day: str condition_night: str humidity: int wind_speed: float wind_dir: str class WeatherForecastDailyResult(BaseModel): location: str forecasts: list[WeatherDailyForecastItem] raw: dict[str, Any] def map_qweather_daily_to_result( *, location: str, raw: dict[str, Any], ) -> WeatherForecastDailyResult: """Map QWeather `/v7/weather/{days}` daily payload into MCP output.""" daily_list: list[dict[str, Any]] = raw.get("daily") or [] forecasts: list[WeatherDailyForecastItem] = [] for d in daily_list: forecasts.append( WeatherDailyForecastItem( date=str(d.get("fxDate")), temp_max=float(d.get("tempMax")), temp_min=float(d.get("tempMin")), condition_day=str(d.get("textDay")), condition_night=str(d.get("textNight")), humidity=int(d.get("humidity")), wind_speed=float(d.get("windSpeedDay")), wind_dir=str(d.get("windDirDay")), ) ) return WeatherForecastDailyResult(location=location, forecasts=forecasts, raw=raw) class WeatherForecastHourlyInput(BaseModel): location: str = Field(min_length=1, max_length=128) hours: Literal[24, 72, 168] = 24 lang: str = "zh" unit: Literal["m", "i"] = "m" class WeatherHourlyForecastItem(BaseModel): time: datetime temperature: float condition: str wind_speed: float wind_dir: str humidity: int class WeatherForecastHourlyResult(BaseModel): location: str hourly: list[WeatherHourlyForecastItem] raw: dict[str, Any] def map_qweather_hourly_to_result( *, location: str, raw: dict[str, Any], ) -> WeatherForecastHourlyResult: """Map QWeather `/v7/weather/{hours}` hourly payload into MCP output.""" hourly_list: list[dict[str, Any]] = raw.get("hourly") or [] items: list[WeatherHourlyForecastItem] = [] for h in hourly_list: items.append( WeatherHourlyForecastItem( time=datetime.fromisoformat(str(h.get("fxTime"))), temperature=float(h.get("temp")), condition=str(h.get("text")), wind_speed=float(h.get("windSpeed")), wind_dir=str(h.get("windDir")), humidity=int(h.get("humidity")), ) ) return WeatherForecastHourlyResult(location=location, hourly=items, raw=raw) class AirQualityCurrentInput(BaseModel): location: str = Field(min_length=1, max_length=128) class AirQualityCurrentResult(BaseModel): location: str aqi: float | None = None category: str | None = None primary_pollutant: str | None = None pm2p5: float | None = None pm10: float | None = None o3: float | None = None no2: float | None = None so2: float | None = None co: float | None = None raw: dict[str, Any] def _find_air_index(raw: dict[str, Any]) -> dict[str, Any] | None: indexes: list[dict[str, Any]] = raw.get("indexes") or [] if not indexes: return None for idx in indexes: if str(idx.get("code")) == "us-epa": return idx return indexes[0] def _pollutant_values(raw: dict[str, Any]) -> dict[str, float]: values: dict[str, float] = {} pollutants: list[dict[str, Any]] = raw.get("pollutants") or [] for p in pollutants: code = str(p.get("code")) conc = p.get("concentration") or {} if "value" in conc: try: values[code] = float(conc.get("value")) except (TypeError, ValueError): continue return values def map_qweather_air_to_result(*, location: str, raw: dict[str, Any]) -> AirQualityCurrentResult: """Map QWeather Air Quality v1 payload into MCP output. Notes: - This API is different from v7 weather endpoints, so we map from `indexes` + `pollutants`. - Pollutant units vary by pollutant and region (see QWeather docs); we preserve raw for detail. """ idx = _find_air_index(raw) or {} primary = idx.get("primaryPollutant") or {} pollutant_map = _pollutant_values(raw) return AirQualityCurrentResult( location=location, aqi=float(idx.get("aqi")) if idx.get("aqi") is not None else None, category=str(idx.get("category")) if idx.get( "category") is not None else None, primary_pollutant=str(primary.get( "name") or primary.get("code")) if primary else None, pm2p5=pollutant_map.get("pm2p5"), pm10=pollutant_map.get("pm10"), o3=pollutant_map.get("o3"), no2=pollutant_map.get("no2"), so2=pollutant_map.get("so2"), co=pollutant_map.get("co"), raw=raw, ) class WeatherAlertsInput(BaseModel): location: str = Field(min_length=1, max_length=128) lang: str = "zh" class WeatherAlertItem(BaseModel): type: str level: str | None = None title: str description: str start: datetime | None = None end: datetime | None = None class WeatherAlertsResult(BaseModel): location: str alerts: list[WeatherAlertItem] raw: dict[str, Any] def map_qweather_warning_to_result(*, location: str, raw: dict[str, Any]) -> WeatherAlertsResult: warning_list: list[dict[str, Any]] = raw.get("warning") or [] alerts: list[WeatherAlertItem] = [] for w in warning_list: level = w.get("severityColor") or w.get("severity") or w.get("level") alerts.append( WeatherAlertItem( type=str(w.get("typeName") or w.get("type")), level=str(level) if level is not None else None, title=str(w.get("title")), description=str(w.get("text")), start=datetime.fromisoformat( str(w.get("startTime"))) if w.get("startTime") else None, end=datetime.fromisoformat( str(w.get("endTime"))) if w.get("endTime") else None, ) ) return WeatherAlertsResult(location=location, alerts=alerts, raw=raw)