Skip to content

src.tools.implementations.weather_tool.WeatherTool

Bases: BaseRESTTool

Source code in src/tools/implementations/weather_tool.py 
 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
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
class WeatherTool(BaseRESTTool):
    name = "weather"

    def __init__(self, config: Optional[Dict] = None):
        super().__init__(config=config)
        self.description = 'Get current temperature and weather information for a specified location.'
        self.strict = False

        self.parameters = {
            'type': 'object',
            'properties': {
                'city': {
                    'type': 'string',
                    'description': 'City name (e.g., "London", "New York")'
                },
                'state_code': {
                    'type': 'string',
                    'description': 'US state code (e.g., "NY", "CA"). Only applicable for US locations.'
                },
                'country_code': {
                    'type': 'string',
                    'description': 'Two-letter country code (e.g., "US", "GB"). Optional, helps disambiguate cities.'
                },
                'units': {
                    'type': 'string',
                    'description': 'Temperature units: Use "metric" for Celsius, "imperial" for Fahrenheit, or "standard" for Kelvin. Defaults to metric.'
                },
                'lang': {
                    'type': 'string',
                    'description': 'Language for weather descriptions (e.g., "en", "es", "fr"). Defaults to "en".'
                }
            },
            'required': ['city'],
            'additionalProperties': False
        }

        self.content_type = "application/json"
        self.rate_limit = 60
        self.default_timeout = 10
        self.max_retries = 3
        self.retry_delay = 1.0

    async def execute(self, context: Optional[StreamContext] = None, **kwargs) -> ToolResponse:
        """Execute a weather data request.

        Args:
            context (Optional[Dict]): Additional context for the request.
            **kwargs: Must include:

                - city (str): City name
                Optional:
                - state_code (str): US state code
                - country_code (str): Two-letter country code
                - units (str): 'metric', 'imperial', or 'standard'
                - lang (str): Language code

        Returns:
            str: Formatted weather information.

        Raises:
            ValueError: If required parameters or API key are missing.
        """
        # Extract and validate parameters
        city = kwargs.get('city')
        if not city:
            raise ValueError("City parameter is required")

        # Get API key from environment
        api_key = os.getenv(self.api_key_env)
        if not api_key:
            raise ValueError(f"API key not found in environment variable: {self.api_key_env}")

        state_code = kwargs.get('state_code')
        country_code = kwargs.get('country_code')
        units = kwargs.get('units', 'metric')
        lang = kwargs.get('lang', 'en')

        # Construct the location query
        location_parts = [city]
        if state_code and country_code == 'US':
            location_parts.append(state_code)
        if country_code:
            location_parts.append(country_code)

        location = ','.join(location_parts)

        # Prepare query parameters
        query_params = {
            'q': location,
            'units': units,
            'lang': lang,
            'appid': api_key  # OpenWeatherMap expects the API key as 'appid'
        }

        # Log the request details
        self.logger.info(f"Making weather request for location: {location}")
        self.logger.info(f"Query parameters: {query_params}")
        self.logger.debug(f"Context: {context}")

        # Make the API request using parent class method
        response = await self.make_request(
            method="GET",
            params=query_params,
            endpoint_url=self.endpoint,
            response_format=ResponseFormat.JSON,
            use_token=False,  # OpenWeatherMap uses an API key
            additional_headers={
                "Accept": "application/json",
                "User-Agent": "WeatherTool/1.0"
            }
        )
        response = ToolResponse(
            result=self.parse_output(response),
            context=None,
        )
        return response

    def parse_output(self, output: Any) -> str:
        """Parse and format the weather API response.

        Args:
            output (Any): Raw API response data.

        Returns:
            str: Formatted weather information including location,
                temperature, humidity, wind, etc.

        Note:
            Handles error responses and includes unit-specific information
            in the output.
        """
        try:
            if not isinstance(output, dict):
                return str(output)

            if 'error' in output:
                return f"Error: {output['error']}"

            if output.get('cod') != 200:
                error_msg = output.get('message', 'Unknown error')
                return f"Error: {error_msg}"

            # Format the weather data into a more readable structure
            weather_info = {
                'location': {
                    'city': output.get('name'),
                    'country': output.get('sys', {}).get('country'),
                    'coordinates': {
                        'latitude': output.get('coord', {}).get('lat'),
                        'longitude': output.get('coord', {}).get('lon')
                    }
                },
                'current_weather': {
                    'temperature': output.get('main', {}).get('temp'),
                    'feels_like': output.get('main', {}).get('feels_like'),
                    'humidity': output.get('main', {}).get('humidity'),
                    'pressure': output.get('main', {}).get('pressure'),
                    'description': output.get('weather', [{}])[0].get('description'),
                    'main': output.get('weather', [{}])[0].get('main'),
                    'wind': {
                        'speed': output.get('wind', {}).get('speed'),
                        'direction': output.get('wind', {}).get('deg')
                    },
                    'clouds': output.get('clouds', {}).get('all'),
                    'visibility': output.get('visibility')
                }
            }

            formatted_output = format_json_to_document(weather_info)
            formatted_output += self.get_tool_specific_instruction()

            return formatted_output

        except Exception as e:
            self.logger.error(f"Failed to parse weather data: {e}", exc_info=True)
            return "An error occurred while parsing the weather information."

    def get_tool_specific_instruction(self) -> str:
        """Get tool-specific instructions about weather data units.

        Returns:
            str: Instructions about weather data units and measurements.
        """
        return (
            "\n\n"
            "## Weather Information Notes: ##\n"
            "- Temperature and feels_like are in the requested units (Celsius for metric, Fahrenheit for imperial)\n"
            "- Wind speed is in meters/sec for metric, miles/hour for imperial\n"
            "- Visibility is in meters\n"
            "- Pressure is in hPa (hectopascals)"
        )

execute(context=None, **kwargs) async

Execute a weather data request.

Parameters:

Name Type Description Default
context Optional[Dict]

Additional context for the request.

 None
**kwargs

Must include:

  • city (str): City name Optional:
  • state_code (str): US state code
  • country_code (str): Two-letter country code
  • units (str): 'metric', 'imperial', or 'standard'
  • lang (str): Language code
 {}

Returns:

Name Type Description
str ToolResponse

Formatted weather information.

Raises:

Type Description
ValueError

If required parameters or API key are missing.
Source code in src/tools/implementations/weather_tool.py 
 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
123
124
125
126
127
async def execute(self, context: Optional[StreamContext] = None, **kwargs) -> ToolResponse:
    """Execute a weather data request.

    Args:
        context (Optional[Dict]): Additional context for the request.
        **kwargs: Must include:

            - city (str): City name
            Optional:
            - state_code (str): US state code
            - country_code (str): Two-letter country code
            - units (str): 'metric', 'imperial', or 'standard'
            - lang (str): Language code

    Returns:
        str: Formatted weather information.

    Raises:
        ValueError: If required parameters or API key are missing.
    """
    # Extract and validate parameters
    city = kwargs.get('city')
    if not city:
        raise ValueError("City parameter is required")

    # Get API key from environment
    api_key = os.getenv(self.api_key_env)
    if not api_key:
        raise ValueError(f"API key not found in environment variable: {self.api_key_env}")

    state_code = kwargs.get('state_code')
    country_code = kwargs.get('country_code')
    units = kwargs.get('units', 'metric')
    lang = kwargs.get('lang', 'en')

    # Construct the location query
    location_parts = [city]
    if state_code and country_code == 'US':
        location_parts.append(state_code)
    if country_code:
        location_parts.append(country_code)

    location = ','.join(location_parts)

    # Prepare query parameters
    query_params = {
        'q': location,
        'units': units,
        'lang': lang,
        'appid': api_key  # OpenWeatherMap expects the API key as 'appid'
    }

    # Log the request details
    self.logger.info(f"Making weather request for location: {location}")
    self.logger.info(f"Query parameters: {query_params}")
    self.logger.debug(f"Context: {context}")

    # Make the API request using parent class method
    response = await self.make_request(
        method="GET",
        params=query_params,
        endpoint_url=self.endpoint,
        response_format=ResponseFormat.JSON,
        use_token=False,  # OpenWeatherMap uses an API key
        additional_headers={
            "Accept": "application/json",
            "User-Agent": "WeatherTool/1.0"
        }
    )
    response = ToolResponse(
        result=self.parse_output(response),
        context=None,
    )
    return response

get_tool_specific_instruction()

Get tool-specific instructions about weather data units.

Returns:

Name Type Description
str str

Instructions about weather data units and measurements.
Source code in src/tools/implementations/weather_tool.py 
189
190
191
192
193
194
195
196
197
198
199
200
201
202
def get_tool_specific_instruction(self) -> str:
    """Get tool-specific instructions about weather data units.

    Returns:
        str: Instructions about weather data units and measurements.
    """
    return (
        "\n\n"
        "## Weather Information Notes: ##\n"
        "- Temperature and feels_like are in the requested units (Celsius for metric, Fahrenheit for imperial)\n"
        "- Wind speed is in meters/sec for metric, miles/hour for imperial\n"
        "- Visibility is in meters\n"
        "- Pressure is in hPa (hectopascals)"
    )

parse_output(output)

Parse and format the weather API response.

Parameters:

Name Type Description Default
output Any

Raw API response data.

 required

Returns:

Name Type Description
str str

Formatted weather information including location, temperature, humidity, wind, etc.
Note

Handles error responses and includes unit-specific information in the output.

Source code in src/tools/implementations/weather_tool.py 
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
def parse_output(self, output: Any) -> str:
    """Parse and format the weather API response.

    Args:
        output (Any): Raw API response data.

    Returns:
        str: Formatted weather information including location,
            temperature, humidity, wind, etc.

    Note:
        Handles error responses and includes unit-specific information
        in the output.
    """
    try:
        if not isinstance(output, dict):
            return str(output)

        if 'error' in output:
            return f"Error: {output['error']}"

        if output.get('cod') != 200:
            error_msg = output.get('message', 'Unknown error')
            return f"Error: {error_msg}"

        # Format the weather data into a more readable structure
        weather_info = {
            'location': {
                'city': output.get('name'),
                'country': output.get('sys', {}).get('country'),
                'coordinates': {
                    'latitude': output.get('coord', {}).get('lat'),
                    'longitude': output.get('coord', {}).get('lon')
                }
            },
            'current_weather': {
                'temperature': output.get('main', {}).get('temp'),
                'feels_like': output.get('main', {}).get('feels_like'),
                'humidity': output.get('main', {}).get('humidity'),
                'pressure': output.get('main', {}).get('pressure'),
                'description': output.get('weather', [{}])[0].get('description'),
                'main': output.get('weather', [{}])[0].get('main'),
                'wind': {
                    'speed': output.get('wind', {}).get('speed'),
                    'direction': output.get('wind', {}).get('deg')
                },
                'clouds': output.get('clouds', {}).get('all'),
                'visibility': output.get('visibility')
            }
        }

        formatted_output = format_json_to_document(weather_info)
        formatted_output += self.get_tool_specific_instruction()

        return formatted_output

    except Exception as e:
        self.logger.error(f"Failed to parse weather data: {e}", exc_info=True)
        return "An error occurred while parsing the weather information."