1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
|
from datetime import UTC, datetime, timedelta, timezone
import requests
from personal_agent.tools.base import Tool, ToolError, ToolResult
class WeatherTool(Tool):
@property
def name(self) -> str:
return "weather"
@property
def description(self) -> str:
return "Get current weather conditions for a location"
@property
def schema(self) -> dict[str, object]:
return {
"type": "object",
"properties": {
"location": {
"type": "string",
"description": ("The city or location to get weather for."),
},
},
"required": ["location"],
}
def execute(self, arguments: dict[str, object]) -> ToolResult:
location = arguments.get("location")
if not isinstance(location, str):
raise ToolError("location must be a string")
geocoding_response = requests.get(
"https://geocoding-api.open-meteo.com/v1/search",
params={
"name": location,
"count": 1,
"language": "en",
"format": "json",
},
timeout=10,
)
geocoding_response.raise_for_status()
geocoding_data = geocoding_response.json()
results = geocoding_data.get("results")
if not results:
raise ToolError(f"location not found: {location}")
place = results[0]
latitude = place["latitude"]
longitude = place["longitude"]
timezone_name = place["timezone"]
forecast_response = requests.get(
"https://api.open-meteo.com/v1/forecast",
params={
"latitude": latitude,
"longitude": longitude,
"hourly": (
"temperature_2m,"
"precipitation_probability,"
"precipitation,"
"weather_code,"
"wind_speed_10m"
),
"timezone": timezone_name,
"forecast_days": 2,
},
timeout=10,
)
forecast_response.raise_for_status()
forecast = forecast_response.json()
hourly = forecast["hourly"]
tomorrow = (
(datetime.now(UTC) + timedelta(days=1))
.astimezone(timezone(timedelta(seconds=forecast["utc_offset_seconds"])))
.date()
)
temperatures = []
precipitation_probabilities = []
precipitation = []
wind_speeds = []
weather_codes = []
for index, timestamp in enumerate(hourly["time"]):
forecast_date = datetime.fromisoformat(timestamp).date()
if forecast_date != tomorrow:
continue
temperatures.append(hourly["temperature_2m"][index])
precipitation_probabilities.append(
hourly["precipitation_probability"][index]
)
precipitation.append(hourly["precipitation"][index])
wind_speeds.append(hourly["wind_speed_10m"][index])
weather_codes.append(hourly["weather_code"][index])
if not temperatures:
raise ToolError("no forecast data available for tomorrow")
return ToolResult(
content=(
f"Location: {place['name']}, {place['country']}\n"
f"Forecast date: {tomorrow}\n"
f"Temperature: {min(temperatures):.1f}–"
f"{max(temperatures):.1f} °C\n"
f"Maximum precipitation probability: "
f"{max(precipitation_probabilities)}%\n"
f"Total precipitation: {sum(precipitation):.1f} mm\n"
f"Maximum wind speed: {max(wind_speeds):.1f} km/h\n"
f"Weather codes: {sorted(set(weather_codes))}"
)
)
|
