get_forecast
Retrieve a 5-day weather forecast for any location by providing latitude and longitude. Uses cached data when available, refreshing every 60 minutes for updated results.
Instructions
Get weather forecast for geographic coordinates.
Provides a 5-day forecast for the specified location. Checks cache first (60 min expiry), fetches fresh if needed.
Args: latitude: Latitude of the location (-90 to 90) longitude: Longitude of the location (-180 to 180)
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| latitude | Yes | ||
| longitude | Yes |
Output Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| result | Yes |
Implementation Reference
- src/mcp_server/server.py:63-89 (handler)MCP tool registration handler for get_forecast. This is the @mcp.tool()-decorated function that serves as the entry point. It caches results using conversation_tools (60 min expiry) and delegates to weather_tools.get_forecast().
@mcp.tool() async def get_forecast(latitude: float, longitude: float) -> str: """Get weather forecast for geographic coordinates. Provides a 5-day forecast for the specified location. Checks cache first (60 min expiry), fetches fresh if needed. Args: latitude: Latitude of the location (-90 to 90) longitude: Longitude of the location (-180 to 180) """ # Check cache first cache_key = f"forecast_{latitude:.2f}_{longitude:.2f}" cached: str | None = conversation_tools.get_cached_weather(cache_key, max_age_minutes=60) if cached: conversation_tools.log_message( "system", f"Returned cached forecast for {latitude}, {longitude}" ) return cached # Fetch fresh data result: str = await weather_tools.get_forecast(latitude, longitude) conversation_tools.cache_weather_data(cache_key, result, "forecast") conversation_tools.log_message( "system", f"Fetched fresh forecast for {latitude}, {longitude}" ) return result - src/mcp_server/server.py:40-89 (registration)Registration of get_forecast via @mcp.tool() decorator inside create_server(). The tool is registered with name 'get_forecast' and takes latitude (float) and longitude (float) as parameters.
@mcp.tool() async def get_alerts(state: str) -> str: """Get active weather alerts for a US state. Provides current weather alerts and warnings for the specified state. Checks cache first (30 min expiry), fetches fresh if needed. Args: state: Two-letter US state code (e.g. CA, NY) """ # Check cache first cache_key = f"alerts_{state.upper()}" cached: str | None = conversation_tools.get_cached_weather(cache_key, max_age_minutes=30) if cached: conversation_tools.log_message("system", f"Returned cached alerts for {state}") return cached # Fetch fresh data result: str = await weather_tools.get_alerts(state) conversation_tools.cache_weather_data(cache_key, result, "alerts") conversation_tools.log_message("system", f"Fetched fresh alerts for {state}") return result @mcp.tool() async def get_forecast(latitude: float, longitude: float) -> str: """Get weather forecast for geographic coordinates. Provides a 5-day forecast for the specified location. Checks cache first (60 min expiry), fetches fresh if needed. Args: latitude: Latitude of the location (-90 to 90) longitude: Longitude of the location (-180 to 180) """ # Check cache first cache_key = f"forecast_{latitude:.2f}_{longitude:.2f}" cached: str | None = conversation_tools.get_cached_weather(cache_key, max_age_minutes=60) if cached: conversation_tools.log_message( "system", f"Returned cached forecast for {latitude}, {longitude}" ) return cached # Fetch fresh data result: str = await weather_tools.get_forecast(latitude, longitude) conversation_tools.cache_weather_data(cache_key, result, "forecast") conversation_tools.log_message( "system", f"Fetched fresh forecast for {latitude}, {longitude}" ) return result - WeatherTools.get_forecast() - the core business logic. Calls the NWS API client, parses 'properties.periods' from the response, formats the first 5 periods with temperature, wind, and forecast details.
async def get_forecast(self, latitude: float, longitude: float) -> str: """Get weather forecast for a location. Args: latitude: Latitude of the location longitude: Longitude of the location Returns: Formatted forecast information or error message """ forecast_data = await self.api_client.get_forecast(latitude, longitude) if not forecast_data: return "Unable to fetch forecast data for this location." try: periods = forecast_data["properties"]["periods"] except (KeyError, TypeError): return "Unable to parse forecast data." # Format the periods into a readable forecast forecasts = [] for period in periods[:5]: # Only show next 5 periods forecast = f""" {period["name"]}: Temperature: {period["temperature"]}°{period["temperatureUnit"]} Wind: {period["windSpeed"]} {period["windDirection"]} Forecast: {period["detailedForecast"]} """ forecasts.append(forecast) return "\n---\n".join(forecasts) - NWSAPIClient.get_forecast() - the low-level HTTP client. First calls the /points/{lat},{lon} endpoint to get the forecast grid URL, then fetches the actual forecast data from the NWS API.
async def get_forecast(self, latitude: float, longitude: float) -> dict[str, Any] | None: """Get weather forecast for coordinates. Args: latitude: Latitude of the location longitude: Longitude of the location Returns: API response data or None if request fails """ # First get the forecast grid endpoint points_url = f"{self.BASE_URL}/points/{latitude},{longitude}" points_data = await self._make_request(points_url) if not points_data: return None # Get the forecast URL from the points response try: forecast_url = points_data["properties"]["forecast"] return await self._make_request(forecast_url) except (KeyError, TypeError): return None - tests/test_tools.py:145-250 (schema)Tests for get_forecast covering success case, no data, invalid data, and period limiting to 5. Tests validate the input/output contract: accepts (latitude, longitude) floats and returns formatted string.
@pytest.mark.asyncio async def test_get_forecast_success(self, weather_tools, mock_api_client): """Test successful forecast retrieval. Arrange: Mock API client to return forecast data Act: Call get_forecast() Assert: Returns formatted forecast """ # Arrange mock_api_client.get_forecast.return_value = { "properties": { "periods": [ { "name": "Tonight", "temperature": 45, "temperatureUnit": "F", "windSpeed": "10 mph", "windDirection": "NW", "detailedForecast": "Clear skies", }, { "name": "Tomorrow", "temperature": 65, "temperatureUnit": "F", "windSpeed": "5 mph", "windDirection": "N", "detailedForecast": "Sunny", }, ] } } # Act result = await weather_tools.get_forecast(37.7749, -122.4194) # Assert assert "Tonight" in result assert "45°F" in result assert "Clear skies" in result assert "Tomorrow" in result mock_api_client.get_forecast.assert_called_once_with(37.7749, -122.4194) @pytest.mark.asyncio async def test_get_forecast_no_data(self, weather_tools, mock_api_client): """Test forecast retrieval with no data. Arrange: Mock API client to return None Act: Call get_forecast() Assert: Returns error message """ # Arrange mock_api_client.get_forecast.return_value = None # Act result = await weather_tools.get_forecast(37.7749, -122.4194) # Assert assert "Unable to fetch forecast" in result @pytest.mark.asyncio async def test_get_forecast_invalid_data(self, weather_tools, mock_api_client): """Test forecast retrieval with invalid data structure. Arrange: Mock API client to return malformed data Act: Call get_forecast() Assert: Returns error message """ # Arrange mock_api_client.get_forecast.return_value = {"invalid": "data"} # Act result = await weather_tools.get_forecast(37.7749, -122.4194) # Assert assert "Unable to parse forecast" in result @pytest.mark.asyncio async def test_get_forecast_limits_periods(self, weather_tools, mock_api_client): """Test that forecast limits to 5 periods. Arrange: Mock API client to return 10 periods Act: Call get_forecast() Assert: Result contains only 5 periods """ # Arrange periods = [ { "name": f"Period {i}", "temperature": 50 + i, "temperatureUnit": "F", "windSpeed": "10 mph", "windDirection": "N", "detailedForecast": f"Forecast {i}", } for i in range(10) ] mock_api_client.get_forecast.return_value = {"properties": {"periods": periods}} # Act result = await weather_tools.get_forecast(37.7749, -122.4194) # Assert assert "Period 0" in result assert "Period 4" in result assert "Period 5" not in result assert result.count("---") == 4 # 5 periods separated by 4 dividers