MCP Weather Server

Instructions

Implementation Reference

• mcp_weather/weather.py:49-130 (handler)

  The handler function for 'get_hourly_weather' tool. It uses AccuWeather API to search for location (with caching), fetch current conditions, and 12-hour hourly forecast. Formats and returns structured data including current and hourly info.

  @mcp.tool()
  async def get_hourly_weather(location: str) -> Dict:
      """Get hourly weather forecast for a location."""
      api_key = os.getenv("ACCUWEATHER_API_KEY")
      base_url = "http://dataservice.accuweather.com"
      
      # Try to get location key from cache first
      location_key = get_cached_location_key(location)
      
      async with ClientSession() as session:
          if not location_key:
              location_search_url = f"{base_url}/locations/v1/cities/search"
              params = {
                  "apikey": api_key,
                  "q": location,
              }
              async with session.get(location_search_url, params=params) as response:
                  locations = await response.json()
                  if response.status != 200:
                      raise Exception(f"Error fetching location data: {response.status}, {locations}")
                  if not locations or len(locations) == 0:
                      raise Exception("Location not found")
              
              location_key = locations[0]["Key"]
              # Cache the location key for future use
              cache_location_key(location, location_key)
          
          # Get current conditions
          current_conditions_url = f"{base_url}/currentconditions/v1/{location_key}"
          params = {
              "apikey": api_key,
          }
          async with session.get(current_conditions_url, params=params) as response:
              current_conditions = await response.json()
              
          # Get hourly forecast
          forecast_url = f"{base_url}/forecasts/v1/hourly/12hour/{location_key}"
          params = {
              "apikey": api_key,
              "metric": "true",
          }
          async with session.get(forecast_url, params=params) as response:
              forecast = await response.json()
          
          # Format response
          hourly_data = []
          for i, hour in enumerate(forecast, 1):
              hourly_data.append({
                  "relative_time": f"+{i} hour{'s' if i > 1 else ''}",
                  "temperature": {
                      "value": hour["Temperature"]["Value"],
                      "unit": hour["Temperature"]["Unit"]
                  },
                  "weather_text": hour["IconPhrase"],
                  "precipitation_probability": hour["PrecipitationProbability"],
                  "precipitation_type": hour.get("PrecipitationType"),
                  "precipitation_intensity": hour.get("PrecipitationIntensity"),
              })
          
          # Format current conditions
          if current_conditions and len(current_conditions) > 0:
              current = current_conditions[0]
              current_data = {
                  "temperature": {
                      "value": current["Temperature"]["Metric"]["Value"],
                      "unit": current["Temperature"]["Metric"]["Unit"]
                  },
                  "weather_text": current["WeatherText"],
                  "relative_humidity": current.get("RelativeHumidity"),
                  "precipitation": current.get("HasPrecipitation", False),
                  "observation_time": current["LocalObservationDateTime"]
              }
          else:
              current_data = "No current conditions available"
          
          return {
              "location": locations[0]["LocalizedName"],
              "location_key": location_key,
              "country": locations[0]["Country"]["LocalizedName"],
              "current_conditions": current_data,
              "hourly_forecast": hourly_data
          } 

• mcp_weather/weather.py:49-49 (registration)

  The @mcp.tool() decorator registers the get_hourly_weather function as an MCP tool.

  @mcp.tool()

• mcp_weather/weather.py:19-29 (helper)

  Helper function to retrieve cached AccuWeather location key for a given location string.

  def get_cached_location_key(location: str) -> Optional[str]:
      """Get location key from cache."""
      if not LOCATION_CACHE_FILE.exists():
          return None
      
      try:
          with open(LOCATION_CACHE_FILE, "r") as f:
              cache = json.load(f)
              return cache.get(location)
      except (json.JSONDecodeError, FileNotFoundError):
          return None

• mcp_weather/weather.py:31-47 (helper)

  Helper function to cache AccuWeather location key for a given location string in JSON file.

  def cache_location_key(location: str, location_key: str):
      """Cache location key for future use."""
      CACHE_DIR.mkdir(parents=True, exist_ok=True)
      
      try:
          if LOCATION_CACHE_FILE.exists():
              with open(LOCATION_CACHE_FILE, "r") as f:
                  cache = json.load(f)
          else:
              cache = {}
          
          cache[location] = location_key
          
          with open(LOCATION_CACHE_FILE, "w") as f:
              json.dump(cache, f, indent=2)
      except Exception as e:
          print(f"Warning: Failed to cache location key: {e}")

• mcp_weather/weather.py:50-51 (schema)

  Function signature and docstring defining input (location: str) and output (Dict), serving as the tool schema.

  async def get_hourly_weather(location: str) -> Dict:
      """Get hourly weather forecast for a location."""