Weather MCP Server

Instructions

Implementation Reference

• src/mcp_weather_server/tools/tools_weather.py:46-85 (handler)

  The run_tool method of GetCurrentWeatherToolHandler that implements the core logic of the get_current_weather tool: validates input, fetches data from WeatherService, formats response, and returns MCP TextContent.

  async def run_tool(self, args: dict) -> Sequence[TextContent | ImageContent | EmbeddedResource]:
      """
      Execute the current weather tool.
      """
      try:
          self.validate_required_args(args, ["city"])
          
          city = args["city"]
          logger.info(f"Getting current weather for: {city}")
          
          # Get weather data from service
          weather_data = await self.weather_service.get_current_weather(city)
          
          # Format the response
          formatted_response = self.weather_service.format_current_weather_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_current_weather: {str(e)}")
          return [
              TextContent(
                  type="text",
                  text=f"Unexpected error occurred: {str(e)}"
              )
          ]

• src/mcp_weather_server/tools/tools_weather.py:25-45 (schema)

  The get_tool_description method defining the Tool metadata including inputSchema for 'city' parameter.

  def get_tool_description(self) -> Tool:
      """
      Return the tool description for current weather lookup.
      """
      return Tool(
          name=self.name,
          description="""Get current weather information for a specified city.
          It extracts the current hour's temperature and weather code, maps
          the weather code to a human-readable description, and returns a formatted summary.""",
          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."
                  }
              },
              "required": ["city"]
          }
      )

• src/mcp_weather_server/server.py:86-86 (registration)

  Registers the GetCurrentWeatherToolHandler instance with the server's tool_handlers registry.

  add_tool_handler(GetCurrentWeatherToolHandler())

• src/mcp_weather_server/tools/weather_service.py:62-140 (helper)

  The get_current_weather method in WeatherService that performs geocoding via get_coordinates, API call to Open-Meteo for hourly forecast, finds current UTC hour, and extracts comprehensive weather variables into a structured dictionary.

  async def get_current_weather(self, city: str) -> Dict[str, Any]:
      """
      Get current weather information for a specified city.
  
      Args:
          city: The name of the city
  
      Returns:
          Dictionary containing current weather data
  
      Raises:
          ValueError: If weather data cannot be retrieved
      """
      try:
          latitude, longitude = await self.get_coordinates(city)
  
          # Build the weather API URL for current conditions 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&forecast_days=1"
          )
  
          logger.info(f"Fetching current weather from: {url}")
  
          async with httpx.AsyncClient() as client:
              weather_response = await client.get(url)
  
              if weather_response.status_code != 200:
                  raise ValueError(f"Weather API returned status {weather_response.status_code}")
  
              weather_data = weather_response.json()
  
              # Find the current hour index
              current_index = utils.get_closest_utc_index(weather_data["hourly"]["time"])
  
              # Extract current weather data with all enhanced variables
              current_weather = {
                  "city": city,
                  "latitude": latitude,
                  "longitude": longitude,
                  "time": weather_data["hourly"]["time"][current_index],
                  "temperature_c": weather_data["hourly"]["temperature_2m"][current_index],
                  "relative_humidity_percent": weather_data["hourly"]["relative_humidity_2m"][current_index],
                  "dew_point_c": weather_data["hourly"]["dew_point_2m"][current_index],
                  "weather_code": weather_data["hourly"]["weather_code"][current_index],
                  "weather_description": utils.weather_descriptions.get(
                      weather_data["hourly"]["weather_code"][current_index],
                      "Unknown weather condition"
                  ),
                  # Wind data
                  "wind_speed_kmh": weather_data["hourly"]["wind_speed_10m"][current_index],
                  "wind_direction_degrees": weather_data["hourly"]["wind_direction_10m"][current_index],
                  "wind_gusts_kmh": weather_data["hourly"]["wind_gusts_10m"][current_index],
                  # Precipitation data
                  "precipitation_mm": weather_data["hourly"]["precipitation"][current_index],
                  "rain_mm": weather_data["hourly"]["rain"][current_index],
                  "snowfall_cm": weather_data["hourly"]["snowfall"][current_index],
                  "precipitation_probability_percent": weather_data["hourly"]["precipitation_probability"][current_index],
                  # Atmospheric data
                  "pressure_hpa": weather_data["hourly"]["pressure_msl"][current_index],
                  "cloud_cover_percent": weather_data["hourly"]["cloud_cover"][current_index],
                  # Comfort & safety
                  "uv_index": weather_data["hourly"]["uv_index"][current_index],
                  "apparent_temperature_c": weather_data["hourly"]["apparent_temperature"][current_index],
                  "visibility_m": weather_data["hourly"]["visibility"][current_index],
              }
  
              return current_weather
  
      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)}")

• src/mcp_weather_server/tools/weather_service.py:228-292 (helper)

  The format_current_weather_response method that transforms the structured weather data into a detailed human-readable text summary, including temperature, feels-like, wind, precipitation, pressure, UV, and visibility.

  def format_current_weather_response(self, weather_data: Dict[str, Any]) -> str:
      """
      Format current weather data into a human-readable string with enhanced variables.
  
      Args:
          weather_data: Weather data dictionary from get_current_weather
  
      Returns:
          Formatted weather description string
      """
      temp = weather_data['temperature_c']
      feels_like = weather_data.get('apparent_temperature_c', temp)
  
      # Temperature text with "feels like" if significantly different
      temp_text = f"temperature of {temp}°C"
      if abs(feels_like - temp) > 2:
          temp_text += f" (feels like {feels_like}°C)"
  
      # Get wind direction as compass direction
      wind_dir = self._degrees_to_compass(weather_data.get('wind_direction_degrees', 0))
  
      # Build the base weather description
      base_text = (
          f"The weather in {weather_data['city']} is {weather_data['weather_description']} "
          f"with a {temp_text}, "
          f"relative humidity at {weather_data['relative_humidity_percent']}%, "
          f"and dew point at {weather_data['dew_point_c']}°C. "
          f"Wind is blowing from the {wind_dir} at {weather_data['wind_speed_kmh']} km/h "
          f"with gusts up to {weather_data['wind_gusts_kmh']} km/h."
      )
  
      # Add precipitation info if present
      precip_mm = weather_data.get('precipitation_mm', 0)
      rain_mm = weather_data.get('rain_mm', 0)
      snow_cm = weather_data.get('snowfall_cm', 0)
      precip_prob = weather_data.get('precipitation_probability_percent', 0)
  
      if precip_mm > 0 or precip_prob > 20:
          if snow_cm > 0:
              base_text += f" Snowfall of {snow_cm} cm is occurring."
          elif rain_mm > 0:
              base_text += f" Rainfall of {rain_mm} mm is occurring."
  
          if precip_prob > 0:
              base_text += f" Precipitation probability is {precip_prob}%."
  
      # Add atmospheric data
      pressure = weather_data.get('pressure_hpa', 0)
      clouds = weather_data.get('cloud_cover_percent', 0)
      base_text += f" Atmospheric pressure is {pressure} hPa with {clouds}% cloud cover."
  
      # Add UV index warning if significant
      uv = weather_data.get('uv_index', 0)
      if uv > 3:
          uv_warning = self._get_uv_warning(uv)
          base_text += f" UV index is {uv:.1f} ({uv_warning})."
  
      # Add visibility
      visibility = weather_data.get('visibility_m', 0)
      if visibility > 0:
          visibility_km = visibility / 1000
          base_text += f" Visibility is {visibility_km:.1f} km."
  
      return base_text