Weather MCP Server

Instructions

Implementation Reference

• main.py:99-113 (handler)

  MCP tool handler dispatch: validates input, calls WeatherAPI.get_forecast, returns JSON-formatted forecast data.

  elif name == "get_weather_forecast":
      location = arguments.get("location", "")
      days = arguments.get("days", 5)
      
      if not location:
          return [TextContent(
              type="text",
              text="Error: location parameter is required"
          )]
      
      forecast_data = await weather_api.get_forecast(location, days)
      return [TextContent(
          type="text",
          text=json.dumps(forecast_data, indent=2)
      )]

• main.py:44-64 (registration)

  Tool registration in list_tools(): defines name, description, and input schema.

  Tool(
      name="get_weather_forecast",
      description="Get weather forecast for a specific location for multiple days",
      inputSchema={
          "type": "object",
          "properties": {
              "location": {
                  "type": "string",
                  "description": "City name or location (e.g., 'New York', 'London')"
              },
              "days": {
                  "type": "integer",
                  "description": "Number of days to forecast (1-5)",
                  "minimum": 1,
                  "maximum": 5,
                  "default": 5
              }
          },
          "required": ["location"]
      }
  ),

• main.py:47-63 (schema)

  Input schema defining parameters: location (required string), days (optional integer 1-5, default 5).

  inputSchema={
      "type": "object",
      "properties": {
          "location": {
              "type": "string",
              "description": "City name or location (e.g., 'New York', 'London')"
          },
          "days": {
              "type": "integer",
              "description": "Number of days to forecast (1-5)",
              "minimum": 1,
              "maximum": 5,
              "default": 5
          }
      },
      "required": ["location"]
  }

• weather_api.py:99-164 (helper)

  WeatherAPI.get_forecast: core logic with caching, OpenWeatherMap API call (or mock), data processing and formatting into structured forecast.

  async def get_forecast(self, location: str, days: int = 5) -> Dict:
      """
      Get weather forecast for a location.
      
      Args:
          location: City name or location string
          days: Number of days to forecast (1-5)
          
      Returns:
          Dictionary with forecast data
      """
      location = self._normalize_location(location)
      days = max(1, min(5, days))  # Clamp between 1 and 5
      
      # Check cache
      cache_key = f"forecast_{location}_{days}"
      cached = self._get_from_cache(cache_key)
      if cached:
          return cached
      
      # If no API key, return mock data
      if not self.api_key:
          return self._get_mock_forecast(location, days)
      
      try:
          async with httpx.AsyncClient() as client:
              url = f"{self.base_url}/forecast"
              params = {
                  "q": location,
                  "appid": self.api_key,
                  "units": "metric",
                  "cnt": days * 8  # 8 forecasts per day (3-hour intervals)
              }
              response = await client.get(url, params=params, timeout=10.0)
              response.raise_for_status()
              data = response.json()
              
              # Format the response
              forecasts = []
              for item in data.get("list", [])[:days * 8]:
                  forecasts.append({
                      "datetime": item.get("dt_txt", ""),
                      "temperature": item.get("main", {}).get("temp", 0),
                      "feels_like": item.get("main", {}).get("feels_like", 0),
                      "humidity": item.get("main", {}).get("humidity", 0),
                      "pressure": item.get("main", {}).get("pressure", 0),
                      "description": item.get("weather", [{}])[0].get("description", ""),
                      "wind_speed": item.get("wind", {}).get("speed", 0),
                      "clouds": item.get("clouds", {}).get("all", 0),
                  })
              
              result = {
                  "location": data.get("city", {}).get("name", location),
                  "country": data.get("city", {}).get("country", ""),
                  "forecasts": forecasts,
                  "days": days,
                  "timestamp": datetime.now().isoformat()
              }
              
              self._cache_data(cache_key, result)
              return result
              
      except httpx.HTTPError as e:
          # Fallback to mock data on API error
          return self._get_mock_forecast(location, days)