MCP Weather SSE Server

Instructions

Args:
    city: City name (e.g. 'London', 'New York')
    days: Number of days (1-5)
    units: Units of measurement ('metric' or 'imperial')

Implementation Reference

• weather.py:259-289 (handler)

  The handler function for the 'get_weather_forecast' tool. It is decorated with @mcp.tool() for registration, includes input schema via type hints and docstring, fetches forecast data from OpenWeatherMap API, handles errors, formats the response using helpers, and returns JSON.

  @mcp.tool()
  async def get_weather_forecast(city: str, days: int = 3, units: str = "metric") -> str:
      """Get weather forecast for a city using OpenWeatherMap.
  
      Args:
          city: City name (e.g. 'London', 'New York')
          days: Number of days (1-5)
          units: Units of measurement ('metric' or 'imperial')
      """
      print(f"get_weather_forecast called with city: {city}, days: {days}, units: {units}", file=sys.stderr)
      
      if not OPENWEATHER_API_KEY:
          return "OpenWeatherMap API key not configured. Please set OPENWEATHER_API_KEY environment variable."
      
      if days < 1 or days > 5:
          return "Days must be between 1 and 5."
      
      url = f"{OPENWEATHER_API_BASE}/forecast"
      params = {
          "q": city,
          "units": units
      }
      
      data = await make_openweather_request(url, params)
      
      if not data:
          return "Unable to fetch forecast data."
      
      result = format_forecast(data, days, units)
      return json.dumps(result, indent=2)

• weather.py:118-172 (helper)

  Helper function that processes and structures the raw forecast data from OpenWeatherMap into a daily grouped dictionary with formatted weather details.

  def format_forecast(data: dict, days: int, units: str) -> dict:
      """Format forecast data from OpenWeatherMap."""
      temp_unit = "°C" if units == "metric" else "°F"
      speed_unit = "m/s" if units == "metric" else "mph"
      
      try:
          city_data = data.get("city", {})
          forecast_list = data.get("list", [])
          
          daily_forecasts = {}
          
          for item in forecast_list:
              date = item.get("dt_txt", "").split(" ")[0]
              
              if date not in daily_forecasts:
                  daily_forecasts[date] = []
              
              daily_forecasts[date].append({
                  "time": item.get("dt_txt", "").split(" ")[1],
                  "temperature": f"{item.get('main', {}).get('temp', 0)}{temp_unit}",
                  "feels_like": f"{item.get('main', {}).get('feels_like', 0)}{temp_unit}",
                  "min_temp": f"{item.get('main', {}).get('temp_min', 0)}{temp_unit}",
                  "max_temp": f"{item.get('main', {}).get('temp_max', 0)}{temp_unit}",
                  "humidity": f"{item.get('main', {}).get('humidity', 0)}%",
                  "pressure": f"{item.get('main', {}).get('pressure', 0)} hPa",
                  "weather": {
                      "main": item.get('weather', [{}])[0].get('main', "Unknown"),
                      "description": item.get('weather', [{}])[0].get('description', "Unknown"),
                      "icon": item.get('weather', [{}])[0].get('icon', "Unknown")
                  },
                  "wind": {
                      "speed": f"{item.get('wind', {}).get('speed', 0)} {speed_unit}",
                      "direction": item.get('wind', {}).get('deg', 0)
                  },
                  "cloudiness": f"{item.get('clouds', {}).get('all', 0)}%"
              })
          
          forecast_dates = list(daily_forecasts.keys())[:days]
          limited_forecasts = {date: daily_forecasts[date] for date in forecast_dates if date in daily_forecasts}
          
          return {
              "location": {
                  "name": city_data.get("name", "Unknown"),
                  "country": city_data.get("country", "Unknown"),
                  "coordinates": {
                      "latitude": city_data.get("coord", {}).get("lat", 0),
                      "longitude": city_data.get("coord", {}).get("lon", 0)
                  }
              },
              "forecast": limited_forecasts
          }
      except Exception as e:
          print(f"Error formatting forecast data: {e}", file=sys.stderr)
          return {"error": "Error formatting forecast data"}

• weather.py:39-55 (helper)

  Helper function to make HTTP requests to the OpenWeatherMap API, adding the API key and handling errors.

  async def make_openweather_request(url: str, params: dict) -> dict[str, Any] | None:
      """Make a request to OpenWeatherMap API with proper error handling."""
      if not OPENWEATHER_API_KEY:
          print("OpenWeatherMap API key not found", file=sys.stderr)
          return None
      
      params["appid"] = OPENWEATHER_API_KEY
      
      async with httpx.AsyncClient() as client:
          try:
              response = await client.get(url, params=params, timeout=30.0)
              response.raise_for_status()
              return response.json()
          except Exception as e:
              print(f"OpenWeatherMap API error: {e}", file=sys.stderr)
              return None