diff options
Diffstat (limited to 'tools.py')
| -rw-r--r-- | tools.py | 233 |
1 files changed, 232 insertions, 1 deletions
@@ -1,3 +1,6 @@ +from datetime import datetime, timedelta, timezone, UTC +from pathlib import Path +import requests from abc import ABC, abstractmethod from dataclasses import dataclass @@ -77,11 +80,12 @@ class HelloTool(Tool): name = arguments.get("name", "stranger") return ToolResult(content=f"Hello, {name}!") + class CalculatorTool(Tool): @property def name(self) -> str: return "calculator" - + @property def description(self) -> str: return "Perform basic arithemtic" @@ -137,6 +141,233 @@ class CalculatorTool(Tool): return ToolResult(content=str(result)) + +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 TypeError("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 ValueError(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 ValueError("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))}" + ) + ) + + +class FileSystemListTool(Tool): + @property + def name(self) -> str: + return "filesystem_list" + + @property + def description(self) -> str: + return "List files and directories inside the agent's sandbox." + + @property + def schema(self) -> dict[str, object]: + return { + "type": "object", + "properties": { + "path": { + "type": "string", + "description": ( + "Relative path inside the filesystem sandbox. " + "Use '.' for the sandbox root." + ), + }, + }, + "required": ["path"], + } + + def execute(self, arguments: dict[str, object]) -> ToolResult: + path = arguments.get("path") + + if not isinstance(path, str): + raise TypeError("path must be a string") + + sandbox = Path.home() / "Projects" / "personal-agent" / "data" + target = (sandbox / path).resolve() + + if not target.is_relative_to(sandbox.resolve()): + raise ValueError("path is outside the filesystem sandbox") + + if not target.exists(): + raise ValueError(f"path does not exist: {path}") + + if not target.is_dir(): + raise ValueError(f"path is not a directory: {path}") + + entries = [] + + for entry in sorted(target.iterdir()): + kind = "directory" if entry.is_dir() else "file" + entries.append(f"{kind}: {entry.name}") + + if not entries: + return ToolResult(content="Directory is empty.") + + return ToolResult(content="\n".join(entries)) + + +class FileSystemReadTool(Tool): + MAX_FILE_SIZE = 1024 * 1024 # 1 MiB + + @property + def name(self) -> str: + return "filesystem_read" + + @property + def description(self) -> str: + return "Read the contents of a file inside the agent's sandbox." + + @property + def schema(self) -> dict[str, object]: + return { + "type": "object", + "properties": { + "path": { + "type": "string", + "description": ("Relative path inside the filesystem sandbox."), + }, + }, + "required": ["path"], + } + + def execute(self, arguments: dict[str, object]) -> ToolResult: + path = arguments.get("path") + + if not isinstance(path, str): + raise TypeError("path must be a string") + + sandbox = Path.home() / "Projects" / "personal-agent" / "data" + sandbox = sandbox.resolve() + target = (sandbox / path).resolve() + + if not target.is_relative_to(sandbox): + raise ValueError("path is outside the filesystem sandbox") + + if not target.exists(): + raise ValueError(f"path does not exist: {path}") + + if not target.is_file(): + raise ValueError(f"path is not a file: {path}") + + if target.stat().st_size > self.MAX_FILE_SIZE: + raise ValueError("file is too large to read") + + try: + content = target.read_text(encoding="utf-8") + except UnicodeDecodeError as error: + raise ValueError("file is not valid UTF-8 text") from error + + return ToolResult(content=content) + + if __name__ == "__main__": registry = ToolRegistry() |