Weather MCP Server

Instructions

Implementation Reference

• src/mcp_weather_server/tools/tools_weather.py:87-167 (handler)

  GetWeatherByDateRangeToolHandler class: the main tool handler implementing run_tool logic, input validation, service call, and response formatting.
  class GetWeatherByDateRangeToolHandler(ToolHandler): """ Tool handler for getting weather information for a date range. """ def __init__(self): super().__init__("get_weather_byDateTimeRange") self.weather_service = WeatherService() def get_tool_description(self) -> Tool: """ Return the tool description for weather date range lookup. """ return Tool( name=self.name, description="""Get weather information for a specified city between start and end dates.""", inputSchema={ "type": "object", "properties": { "city": { "type": "string", "description": "The name of the city to fetch weather information for, PLEASE NOTE English name only, if the parameter city isn't English please translate to English before invoking this function." }, "start_date": { "type": "string", "description": "Start date in format YYYY-MM-DD, please follow ISO 8601 format" }, "end_date": { "type": "string", "description": "End date in format YYYY-MM-DD , please follow ISO 8601 format" } }, "required": ["city", "start_date", "end_date"] } ) async def run_tool(self, args: dict) -> Sequence[TextContent | ImageContent | EmbeddedResource]: """ Execute the weather date range tool. """ try: self.validate_required_args(args, ["city", "start_date", "end_date"]) city = args["city"] start_date = args["start_date"] end_date = args["end_date"] logger.info(f"Getting weather for {city} from {start_date} to {end_date}") # Get weather data from service weather_data = await self.weather_service.get_weather_by_date_range( city, start_date, end_date ) # Format the response for analysis formatted_response = self.weather_service.format_weather_range_response(weather_data) return [ TextContent( type="text", text=formatted_response ) ] except ValueError as e: logger.error(f"Weather service error: {str(e)}") return [ TextContent( type="text", text=f"Error: {str(e)}" ) ] except Exception as e: logger.exception(f"Unexpected error in get_weather_by_date_range: {str(e)}") return [ TextContent( type="text", text=f"Unexpected error occurred: {str(e)}" ) ]
• src/mcp_weather_server/tools/tools_weather.py:96-121 (schema)

  get_tool_description method defining the Tool schema with inputSchema for city (string), start_date (string YYYY-MM-DD), end_date (string YYYY-MM-DD).
  def get_tool_description(self) -> Tool: """ Return the tool description for weather date range lookup. """ return Tool( name=self.name, description="""Get weather information for a specified city between start and end dates.""", inputSchema={ "type": "object", "properties": { "city": { "type": "string", "description": "The name of the city to fetch weather information for, PLEASE NOTE English name only, if the parameter city isn't English please translate to English before invoking this function." }, "start_date": { "type": "string", "description": "Start date in format YYYY-MM-DD, please follow ISO 8601 format" }, "end_date": { "type": "string", "description": "End date in format YYYY-MM-DD , please follow ISO 8601 format" } }, "required": ["city", "start_date", "end_date"] } )
• src/mcp_weather_server/server.py:78-99 (registration)

  register_all_tools function that instantiates and registers the GetWeatherByDateRangeToolHandler along with other tools using add_tool_handler. Called during server startup.
  def register_all_tools() -> None: """ Register all available tool handlers. This function serves as the central registry for all tools. New tool handlers should be added here for automatic registration. """ # Weather tools add_tool_handler(GetCurrentWeatherToolHandler()) add_tool_handler(GetWeatherByDateRangeToolHandler()) add_tool_handler(GetWeatherDetailsToolHandler()) # Time tools add_tool_handler(GetCurrentDateTimeToolHandler()) add_tool_handler(GetTimeZoneInfoToolHandler()) add_tool_handler(ConvertTimeToolHandler()) # Air quality tools add_tool_handler(GetAirQualityToolHandler()) add_tool_handler(GetAirQualityDetailsToolHandler()) logger.info(f"Registered {len(tool_handlers)} tool handlers")
• src/mcp_weather_server/tools/weather_service.py:141-227 (helper)

  WeatherService.get_weather_by_date_range: core helper method that geocodes the city, fetches hourly weather data from Open-Meteo API for the date range, processes into structured data with multiple weather variables.
  async def get_weather_by_date_range( self, city: str, start_date: str, end_date: str ) -> Dict[str, Any]: """ Get weather information for a specified city between start and end dates. Args: city: The name of the city start_date: Start date in YYYY-MM-DD format end_date: End date in YYYY-MM-DD format Returns: Dictionary containing weather data for the date range Raises: ValueError: If weather data cannot be retrieved """ try: latitude, longitude = await self.get_coordinates(city) # Build the weather API URL for date range with enhanced variables url = ( f"{self.BASE_WEATHER_URL}" f"?latitude={latitude}&longitude={longitude}" f"&hourly=temperature_2m,relative_humidity_2m,dew_point_2m,weather_code," f"wind_speed_10m,wind_direction_10m,wind_gusts_10m," f"precipitation,rain,snowfall,precipitation_probability," f"pressure_msl,cloud_cover,uv_index,apparent_temperature,visibility" f"&timezone=GMT&start_date={start_date}&end_date={end_date}" ) logger.info(f"Fetching weather history from: {url}") async with httpx.AsyncClient() as client: response = await client.get(url) if response.status_code != 200: raise ValueError(f"Weather API returned status {response.status_code}") data = response.json() # Process the hourly data with enhanced variables weather_data = [] hourly = data["hourly"] data_length = len(hourly["time"]) for i in range(data_length): weather_data.append({ "time": hourly["time"][i], "temperature_c": hourly["temperature_2m"][i], "humidity_percent": hourly["relative_humidity_2m"][i], "dew_point_c": hourly["dew_point_2m"][i], "weather_code": hourly["weather_code"][i], "weather_description": utils.weather_descriptions.get( hourly["weather_code"][i], "Unknown weather condition" ), "wind_speed_kmh": hourly["wind_speed_10m"][i], "wind_direction_degrees": hourly["wind_direction_10m"][i], "wind_gusts_kmh": hourly["wind_gusts_10m"][i], "precipitation_mm": hourly["precipitation"][i], "rain_mm": hourly["rain"][i], "snowfall_cm": hourly["snowfall"][i], "precipitation_probability_percent": hourly["precipitation_probability"][i], "pressure_hpa": hourly["pressure_msl"][i], "cloud_cover_percent": hourly["cloud_cover"][i], "uv_index": hourly["uv_index"][i], "apparent_temperature_c": hourly["apparent_temperature"][i], "visibility_m": hourly["visibility"][i], }) return { "city": city, "latitude": latitude, "longitude": longitude, "start_date": start_date, "end_date": end_date, "weather_data": weather_data } except httpx.RequestError as e: raise ValueError(f"Network error while fetching weather for {city}: {str(e)}") except (KeyError, IndexError) as e: raise ValueError(f"Invalid response format from weather API: {str(e)}")