get_alerts

Instructions

Args:
    state: Two-letter US state code (e.g. CA, NY)

Implementation Reference

• weather.py:108-168 (handler)

  The main execution logic for the 'get_alerts' tool. Fetches active weather alerts for a given US state using the NWS API, validates input, handles errors, processes multiple alerts with progress reporting, and formats the output using helper functions.

  @mcp.tool()
  async def get_alerts(state: str, ctx: Context[ServerSession, None]) -> str:
      """Get weather alerts for a US state.
  
      Args:
          state: Two-letter US state code (e.g. CA, NY)
      """
      await ctx.info(f"Fetching weather alerts for state: {state}")
      logger.info(f"Processing alerts request for state: {state}")
      
      # Validate state code
      if len(state) != 2:
          error_msg = f"Invalid state code format: {state}. Expected 2-letter code."
          await ctx.warning(error_msg)
          logger.warning(error_msg)
          return "Error: State code must be exactly 2 letters (e.g., CA, NY)"
      
      state = state.upper()
      url = f"{NWS_API_BASE}/alerts/active/area/{state}"
      
      await ctx.debug(f"Requesting alerts from URL: {url}")
      data = await make_nws_request(url, ctx)
  
      if not data:
          error_msg = f"Unable to fetch alerts for state: {state}"
          await ctx.error(error_msg)
          logger.error(error_msg)
          return "Unable to fetch alerts or no alerts found."
  
      if "features" not in data:
          error_msg = f"Invalid response format for state: {state}"
          await ctx.error(error_msg)
          logger.error(error_msg)
          return "Invalid response format from weather service."
  
      if not data["features"]:
          await ctx.info(f"No active alerts found for state: {state}")
          logger.info(f"No active alerts for state: {state}")
          return "No active alerts for this state."
  
      alert_count = len(data["features"])
      await ctx.info(f"Processing {alert_count} alerts for state: {state}")
      logger.info(f"Found {alert_count} alerts for state: {state}")
      
      # Process alerts with progress reporting
      alerts = []
      for i, feature in enumerate(data["features"]):
          progress = (i + 1) / alert_count
          await ctx.report_progress(
              progress=progress,
              total=1.0,
              message=f"Processing alert {i + 1}/{alert_count}"
          )
          
          formatted_alert = await format_alert(feature, ctx)
          alerts.append(formatted_alert)
      
      await ctx.info(f"Successfully processed {alert_count} alerts for state: {state}")
      logger.info(f"Successfully processed {alert_count} alerts for state: {state}")
      
      return "\n---\n".join(alerts)

• weather.py:74-105 (helper)

  Helper function to format an individual weather alert feature into a human-readable string, extracting key properties like event, area, severity, description, and instructions.

  async def format_alert(feature: dict, ctx: Context[ServerSession, None] | None = None) -> str:
      """Format an alert feature into a readable string."""
      try:
          props = feature["properties"]
          alert_event = props.get('event', 'Unknown')
          
          if ctx:
              await ctx.debug(f"Formatting alert for event: {alert_event}")
          logger.debug(f"Formatting alert for event: {alert_event}")
          
          formatted_alert = f"""
  Event: {alert_event}
  Area: {props.get('areaDesc', 'Unknown')}
  Severity: {props.get('severity', 'Unknown')}
  Description: {props.get('description', 'No description available')}
  Instructions: {props.get('instruction', 'No specific instructions provided')}
  """
          return formatted_alert
          
      except KeyError as e:
          error_msg = f"Missing required field in alert data: {e}"
          if ctx:
              await ctx.warning(error_msg)
          logger.warning(error_msg)
          return "Error: Invalid alert data format"
          
      except Exception:
          error_msg = "Unexpected error while formatting alert"
          if ctx:
              await ctx.error(error_msg)
          logger.exception("Unexpected error while formatting alert")
          return "Error: Could not format alert"

• weather.py:26-71 (helper)

  Shared helper function for making HTTP requests to the NWS API, with comprehensive error handling for timeouts, HTTP errors, network issues, and logging via MCP context.

  async def make_nws_request(url: str, ctx: Context[ServerSession, None] | None = None) -> dict[str, Any] | None:
      """Make a request to the NWS API with proper error handling."""
      headers = {"User-Agent": USER_AGENT, "Accept": "application/geo+json"}
      
      if ctx:
          await ctx.debug(f"Making NWS API request to: {url}")
      logger.debug(f"Making request to NWS API: {url}")
      
      async with httpx.AsyncClient(follow_redirects=True) as client:
          try:
              response = await client.get(url, headers=headers, timeout=30.0)
              response.raise_for_status()
              
              if ctx:
                  await ctx.debug(f"NWS API request successful, status: {response.status_code}")
              logger.debug(f"NWS API request successful: {response.status_code}")
              
              return response.json()
              
          except httpx.TimeoutException:
              error_msg = f"Request timeout for URL: {url}"
              if ctx:
                  await ctx.error(error_msg)
              logger.exception("Request timeout occurred")
              return None
              
          except httpx.HTTPStatusError as e:
              error_msg = f"HTTP error {e.response.status_code} for URL: {url}"
              if ctx:
                  await ctx.error(error_msg)
              logger.exception("HTTP status error occurred")
              return None
              
          except httpx.RequestError:
              error_msg = f"Network error for URL: {url}"
              if ctx:
                  await ctx.error(error_msg)
              logger.exception("Network request error occurred")
              return None
              
          except Exception:
              error_msg = f"Unexpected error for URL: {url}"
              if ctx:
                  await ctx.error(error_msg)
              logger.exception("Unexpected error during NWS API request")
              return None