get_water_level_condition
Retrieve flood warning water level conditions for specific districts or states in Malaysia to monitor potential flooding risks.
Instructions
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.
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| district | No | ||
| state | No |
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')} """