"""Weather forecast tool via Open-Meteo (free, no API key)."""
import logging
import httpx
from pydantic import BaseModel, Field
from ..bot import BotContext
from .base import ToolResult
from .registry import folder_bot
logger = logging.getLogger(__name__)
_OPEN_METEO_URL = "https://api.open-meteo.com/v1/forecast"
_WMO_CODES: dict[int, str] = {
0: "Clear sky",
1: "Mainly clear",
2: "Partly cloudy",
3: "Overcast",
45: "Fog",
48: "Depositing rime fog",
51: "Light drizzle",
53: "Moderate drizzle",
55: "Dense drizzle",
56: "Light freezing drizzle",
57: "Dense freezing drizzle",
61: "Light rain",
63: "Moderate rain",
65: "Heavy rain",
66: "Light freezing rain",
67: "Heavy freezing rain",
71: "Light snowfall",
73: "Moderate snowfall",
75: "Heavy snowfall",
77: "Snow grains",
80: "Light rain showers",
81: "Moderate rain showers",
82: "Violent rain showers",
85: "Light snow showers",
86: "Heavy snow showers",
95: "Thunderstorm",
96: "Thunderstorm with light hail",
99: "Thunderstorm with heavy hail",
}
def _describe_weather_code(code: int) -> str:
"""Convert a WMO weather code to a human-readable description."""
return _WMO_CODES.get(code, f"Unknown ({code})")
[docs]
class GetWeatherRequest(BaseModel, frozen=True):
"""Request for getting weather forecast."""
latitude: float = Field(description="Latitude coordinate")
longitude: float = Field(description="Longitude coordinate")
forecast_days: int = Field(
default=3,
description="Number of forecast days (1-7)",
ge=1,
le=7,
)
[docs]
@folder_bot.tool(
name="get_weather",
request_type=GetWeatherRequest,
response_type=ToolResult,
)
async def get_weather(
request: GetWeatherRequest, _context: BotContext | None = None
) -> ToolResult:
"""Get current weather and forecast for a location.
Requires latitude and longitude — use ask_user with input_type 'location'
to get the user's coordinates if not known.
"""
try:
params: dict[str, str | int | float] = {
"latitude": request.latitude,
"longitude": request.longitude,
"current": ",".join(
[
"temperature_2m",
"apparent_temperature",
"relative_humidity_2m",
"weather_code",
"wind_speed_10m",
]
),
"daily": ",".join(
[
"temperature_2m_max",
"temperature_2m_min",
"precipitation_sum",
"weather_code",
]
),
"forecast_days": request.forecast_days,
"timezone": "auto",
}
with httpx.Client(timeout=15.0) as client:
response = client.get(_OPEN_METEO_URL, params=params)
response.raise_for_status()
data = response.json()
# Format current conditions
current = data["current"]
units = data["current_units"]
lines = [
"Current conditions:",
f" {_describe_weather_code(current['weather_code'])}",
f" Temperature: {current['temperature_2m']}{units['temperature_2m']}"
f" (feels like {current['apparent_temperature']}{units['apparent_temperature']})",
f" Humidity: {current['relative_humidity_2m']}{units['relative_humidity_2m']}",
f" Wind: {current['wind_speed_10m']}{units['wind_speed_10m']}",
"",
"Forecast:",
]
# Format daily forecast
daily = data["daily"]
for i, date in enumerate(daily["time"]):
code = daily["weather_code"][i]
hi = daily["temperature_2m_max"][i]
lo = daily["temperature_2m_min"][i]
precip = daily["precipitation_sum"][i]
d_units = data["daily_units"]
line = (
f" {date}: {_describe_weather_code(code)}"
f", {lo}–{hi}{d_units['temperature_2m_max']}"
f", precip {precip}{d_units['precipitation_sum']}"
)
lines.append(line)
return ToolResult(content="\n".join(lines))
except Exception as e:
logger.exception("Weather fetch error")
return ToolResult(content=f"Weather error: {e}", is_error=True)