get_water_level_condition

Instructions

Args:
    district: The name of the district within the specified state for which to retrieve flood warning conditions.
    state: The name of the state in Malaysia for which to retrieve flood warning conditions.

Implementation Reference

• weather.py:10-44 (handler)

  The core handler function implementing the tool logic. It processes input district/state, queries the Malaysian government flood-warning API for active (ON) water level statuses, handles errors/no data, formats results using format_water_level helper, and returns joined summaries.

  async def get_water_level_condition(district: str = "", state: str = "") -> str:
      """Retrieve the water level conditions associated with flood warnings for a specified district or state.
      If both district and state are provided, the district takes precedence.
      If district or state is not specified, use an empty string for that field.
  
      Args:
          district: The name of the district within the specified state for which to retrieve flood warning conditions.
          state: The name of the state in Malaysia for which to retrieve flood warning conditions.
      """
  
      contain_str = ""
      district = district.strip()
      state = state.strip()
      if district:
          contain_str = f"{district}@district"
      elif state:
          contain_str = f"{state}@state"
  
      water_level_url = f"{GOV_API_BASE}/flood-warning"
      water_level_data = await make_api_request(water_level_url, {
                                              "meta": "true",
                                              "sort": "-water_level_update_datetime,-rainfall_update_datetime",
                                              "icontains": contain_str,
                                              "filter": "ON@water_level_status",
                                              "limit": 20
                                          })
  
      if not water_level_data or "data" not in water_level_data:
          return "Unable to fetch water level data for this state or district."
  
      if not water_level_data["data"]:
          return "No active water level data for this state or district."
  
      water_levels = [format_water_level(warn) for warn in water_level_data["data"]]
      return "\n---\n".join(water_levels)

• weather.py:9-9 (registration)

  The @mcp.tool() decorator registers the get_water_level_condition function as a tool on the FastMCP server instance.

  @mcp.tool()

• helpers.py:49-74 (helper)

  Supporting helper function that formats the API response data for water level stations into a detailed human-readable string used in the tool output.

  def format_water_level(water_level_res: dict) -> str:
      """Format water level and rainfail data into a readable string."""
      return f"""
  Monitoring Station ID: {water_level_res.get('station_id', 'Unknown')}
  Monitoring Station Name: {water_level_res.get('station_name', 'Unknown')}
  Latitude: {water_level_res.get('latitude', 'Unknown')}
  Longitude: {water_level_res.get('longitude', 'Unknown')}
  District: {water_level_res.get('district', 'Unknown')}
  State: {water_level_res.get('state', 'Unknown')}
  Sub-basin: {water_level_res.get('sub_basin', 'Unknown')}
  Main River Basin: {water_level_res.get('main_basin', 'Unknown')}
  Current Water Level: {water_level_res.get('water_level_current', 'Unknown')}
  Current Water Level Indicator: {water_level_res.get('water_level_indicator', 'Unknown')}
  Current Water Level Increment: {water_level_res.get('water_level_increment', 'Unknown')}
  Current Water Level Trend (rising, falling, or steady): {water_level_res.get('water_level_trend', 'Unknown')}
  Water Level - Normal (Reference): {water_level_res.get('water_level_normal_level', 'Unknown')}
  Water Level - Alert (Reference): {water_level_res.get('water_level_alert_level', 'Unknown')}
  Water Level - Warning (Reference): {water_level_res.get('water_level_warning_level', 'Unknown')}
  Water Level - Danger (Reference): {water_level_res.get('water_level_danger_level', 'Unknown')}
  Current Water Level Updated Datetime: {water_level_res.get('water_level_update_datetime', 'Unknown')}
  Clean Rainfall: {water_level_res.get('rainfall_clean', 'Unknown')}
  Latest 1 Hour Rainfall: {water_level_res.get('rainfall_latest_1hr', 'Unknown')}
  Total Rainfall Today: {water_level_res.get('rainfall_total_today', 'Unknown')}
  Rainfall Indicator: {water_level_res.get('rainfall_indicator', 'Unknown')}
  Rainfall Updated Datetime: {water_level_res.get('rainfall_update_datetime', 'Unknown')}
  """