Weather MCP MY

Instructions

Implementation Reference

• weather.py:9-44 (handler)

  The handler function for the 'get_water_level_condition' tool. It constructs an API query based on district or state, fetches data from the flood-warning endpoint, and formats the results.
  @mcp.tool() 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)
• helpers.py:49-74 (helper)

  Helper function to format water level response data into a readable multi-line string, used by the tool handler.
  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')} """
• weather.py:9-9 (registration)

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