""" Philippines Disaster Risk Assessment MCP Server An MCP (Model Context Protocol) server that provides comprehensive disaster risk assessment data for any location in the Philippines. This server interfaces with the GeoRisk Philippines API to deliver real-time hazard information including: - Seismic hazards (active faults, ground shaking, liquefaction, tsunami) - Volcanic hazards (active volcanoes, pyroclastic flows, ashfall, lahar) - Hydrometeorological hazards (floods, landslides, storm surge, severe winds) - Critical infrastructure proximity (schools, hospitals, road networks) Features: - Intelligent caching (1-hour TTL) to reduce API load - Input validation for geographic coordinates - Secure authentication with GeoRisk API - User-friendly Gradio web interface Usage: - Call with latitude and longitude coordinates (-90 to 90, -180 to 180) - Returns formatted hazard assessment reports with risk levels and recommendations """ import gradio as gr from src.ph_disaster_risk_mcp.api_client import GeoRiskClient from src.ph_disaster_risk_mcp.auth import AuthManager from src.ph_disaster_risk_mcp.cache import CacheManager from src.ph_disaster_risk_mcp.formatter import format_hazard_data, format_error from src.ph_disaster_risk_mcp.exceptions import GeoRiskError # Initialize global instances for authentication, caching, and API client auth_manager = AuthManager() cache_manager = CacheManager(ttl_seconds=3600) # 1-hour cache TTL api_client = GeoRiskClient(auth_manager=auth_manager) def get_risk_data(latitude: float, longitude: float) -> dict: """ Get comprehensive disaster risk assessment for any location in the Philippines. Provides detailed hazard information from the GeoRisk Philippines API including: - Seismic hazards (active faults, ground shaking, liquefaction, tsunami) - Volcanic hazards (active volcanoes, pyroclastic flows, ashfall, lahar) - Hydrometeorological hazards (floods, landslides, storm surge, severe winds) - Critical infrastructure proximity (schools, hospitals, road networks) Returns structured JSON data with automatic risk categorization, summary statistics, and overall risk level assessment (CRITICAL/HIGH/MODERATE/LOW). Args: latitude: Latitude coordinate (-90 to 90 degrees) longitude: Longitude coordinate (-180 to 180 degrees) Returns: dict: Structured disaster risk assessment containing: - success: Boolean indicating success/failure - summary: Risk overview with total_hazards_assessed, high_risk_count, moderate_risk_count, critical_hazards list, overall_risk_level - location: Coordinate and location name information - hazards: Categorized by type (seismic, volcanic, hydrometeorological) - facilities: Organized by category (schools, hospitals, roads) - metadata: Cache status, timestamp, source, and TTL info Example: get_risk_data(14.5995, 120.9842) # Manila, Philippines Returns: {"success": true, "summary": {...}, "location": {...}, ...} """ try: # Check cache for existing response from_cache = False cached_response = cache_manager.get(latitude, longitude) if cached_response is not None: # Cache hit - use cached data response = cached_response from_cache = True else: # Cache miss - make API call response = api_client.get_hazard_assessment(latitude, longitude) # Store response in cache cache_manager.set(latitude, longitude, response) # Extract and structure the data structured_data = _structure_response(response, latitude, longitude, from_cache) return structured_data except GeoRiskError as e: # Handle all GeoRisk-related errors (validation, auth, API errors) return { "error": True, "error_type": type(e).__name__, "message": str(e), "location": {"latitude": latitude, "longitude": longitude} } except Exception as e: # Handle unexpected errors return { "error": True, "error_type": type(e).__name__, "message": str(e), "location": {"latitude": latitude, "longitude": longitude} } def _structure_response(response: dict, latitude: float, longitude: float, from_cache: bool) -> dict: from datetime import datetime # Extract data section data = response.get("data", response) if isinstance(response, dict) else response # Categorize hazards by severity high_risks = [] moderate_risks = [] low_risks = [] # Define hazard categories for structured output hazard_fields = { "seismic": [ "activeFault", "groundRupture", "groundShaking", "eil", "liquefaction", "fissure", "tsunami" ], "volcanic": [ "activeVolcano", "potentiallyActiveVolcano", "inactiveVolcano", "ballisticProjectile", "baseSurge", "lahar", "lava", "pyroclasticFlow", "volcanicFissure", "volcanicTsunami", "pdz", "edz", "ashfall" ], "hydrometeorological": [ "flood", "ril", "stormSurge", "severeWinds" ] } # Process hazards and categorize by risk level categorized_hazards = {} for category, fields in hazard_fields.items(): categorized_hazards[category] = {} for field in fields: hazard = data.get(field, {}) if hazard and isinstance(hazard, dict): assessment = hazard.get("assessment", "").lower() categorized_hazards[category][field] = hazard # Categorize by risk level hazard_info = {"name": field, "assessment": hazard.get("assessment", "N/A")} if any(word in assessment for word in ["high", "High", "very high", "extreme", "critical", "within", "prone", "highly susceptible", "high susceptibility"]): high_risks.append(hazard_info) elif any(word in assessment for word in ["moderate", "moderately", "medium", "present"]): moderate_risks.append(hazard_info) elif any(word in assessment for word in ["low", "minimal", "none", "safe", "outside"]): low_risks.append(hazard_info) # Process critical facilities facilities = { "schools": { "elementary": data.get("elementarySchool", {}), "secondary": data.get("secondarySchool", {}) }, "hospitals": { "government": data.get("governmentHospital", {}), "private": data.get("privateHospital", {}) }, "roads": { "primary": data.get("primaryRoadNetwork", {}), "secondary": data.get("secondaryRoadNetwork", {}) } } # Build structured response return { "success": True, "summary": { "total_hazards_assessed": sum(len(v) for v in categorized_hazards.values()), "high_risk_count": len(high_risks), "moderate_risk_count": len(moderate_risks), "low_risk_count": len(low_risks), "critical_hazards": high_risks[:5], # Top 5 critical hazards "overall_risk_level": _calculate_overall_risk(len(high_risks), len(moderate_risks)) }, "location": { "latitude": latitude, "longitude": longitude, "name": data.get("location", {}).get("name", "Unknown Location") }, "hazards": categorized_hazards, "facilities": facilities, "metadata": { "from_cache": from_cache, "timestamp": datetime.utcnow().isoformat() + "Z", "source": "GeoRisk Philippines API", "cache_ttl_seconds": 3600 } } def _calculate_overall_risk(high_count: int, moderate_count: int) -> str: if high_count >= 3: return "CRITICAL" elif high_count >= 1: return "HIGH" elif moderate_count >= 3: return "MODERATE" elif moderate_count >= 1: return "LOW-MODERATE" else: return "LOW" def get_risk_data_html(data: dict) -> str: """ Format disaster risk data as HTML for web display. Takes structured JSON data from get_risk_data() and converts it to a beautiful HTML display with color-coded risk summary banner and detailed hazard information organized by category. This function is useful for: - Gradio web UI display - Embedding in web pages - Custom HTML rendering needs For MCP/API usage, use get_risk_data() instead which returns structured JSON. Args: data: Structured risk assessment dict from get_risk_data() Returns: str: HTML formatted string with risk assessment display including: - Color-coded summary banner (red/yellow/green based on risk level) - Overall risk level and hazard counts - Detailed seismic, volcanic, and hydrometeorological hazards - Critical facilities information Example: json_data = get_risk_data(14.5995, 120.9842) html = get_risk_data_html(json_data) """ # Handle errors if data.get("error"): return format_error_display(data) # Use the original formatter by reconstructing the format it expects # This maintains backward compatibility with the existing formatter reconstructed_data = { "data": { "location": {"name": data["location"]["name"]}, "coordinates": { "latitude": data["location"]["latitude"], "longitude": data["location"]["longitude"] } } } # Add all hazards back for category in data["hazards"].values(): for field_name, field_data in category.items(): reconstructed_data["data"][field_name] = field_data # Add facilities back reconstructed_data["data"]["elementarySchool"] = data["facilities"]["schools"]["elementary"] reconstructed_data["data"]["secondarySchool"] = data["facilities"]["schools"]["secondary"] reconstructed_data["data"]["governmentHospital"] = data["facilities"]["hospitals"]["government"] reconstructed_data["data"]["privateHospital"] = data["facilities"]["hospitals"]["private"] reconstructed_data["data"]["primaryRoadNetwork"] = data["facilities"]["roads"]["primary"] reconstructed_data["data"]["secondaryRoadNetwork"] = data["facilities"]["roads"]["secondary"] # Add summary banner at the top summary_html = f"""

⚠️ Overall Risk Level: {data['summary']['overall_risk_level']}

{data['summary']['high_risk_count']} high-risk hazards | {data['summary']['moderate_risk_count']} moderate-risk hazards | {data['summary']['total_hazards_assessed']} total hazards assessed

{('

🚨 Critical: ' + ', '.join([h['name'] for h in data['summary']['critical_hazards'][:3]]) + '

') if data['summary']['critical_hazards'] else ''}
""" return summary_html + format_hazard_data(reconstructed_data) def format_error_display(error_data: dict) -> str: return f"""

❌ Error: {error_data.get('error_type', 'Unknown')}

{error_data.get('message', 'An error occurred')}

""" def create_interface() -> gr.Blocks: with gr.Blocks( title="PH Disaster Risk Assessment", theme=gr.themes.Soft() ) as demo: gr.Markdown("# 🌋 Philippines Disaster Risk Assessment") gr.Markdown( "Get comprehensive disaster risk information for any location in the Philippines. " "This tool provides hazard assessments from the GeoRisk Philippines API, including:\n" "- **Seismic Hazards**: Active faults, ground shaking, liquefaction, tsunami\n" "- **Volcanic Hazards**: Active volcanoes, pyroclastic flows, ashfall, lahar\n" "- **Hydrometeorological Hazards**: Floods, landslides, storm surge, severe winds\n" "- **Critical Facilities**: Nearby schools, hospitals, and road networks\n\n" "**✨ MCP-Optimized**: This server returns structured data for easy LLM consumption " "with summary statistics and risk categorization." ) gr.Markdown("### 📍 Enter Coordinates") gr.Markdown( "Enter the latitude and longitude of the location you want to assess. " "You can use the example coordinates below or enter your own." ) with gr.Row(): latitude_input = gr.Number( label="Latitude", value=14.5995, info="Enter latitude (-90 to 90)", precision=6 ) longitude_input = gr.Number( label="Longitude", value=120.9842, info="Enter longitude (-180 to 180)", precision=6 ) submit_btn = gr.Button("🔍 Get Risk Assessment", variant="primary", size="lg") gr.Markdown("---") json_data = gr.JSON(visible=False) output = gr.HTML(label="Risk Assessment Results") # get_risk_data returns JSON dict (for MCP) # Format it as HTML for UI display submit_btn.click( fn=get_risk_data, inputs=[latitude_input, longitude_input], outputs=json_data ).then( fn=get_risk_data_html, inputs=json_data, outputs=output ) gr.Markdown("### 📌 Example Coordinates") gr.Markdown("Click on any example below to load the coordinates:") gr.Examples( examples=[ [14.5995, 120.9842], # Manila [10.3157, 123.8854], # Cebu City [7.1907, 125.4553], # Davao City [16.4023, 120.5960], # Baguio City [11.2500, 125.0000], # Tacloban City ], inputs=[latitude_input, longitude_input], label="Try these Philippine locations", examples_per_page=5 ) gr.Markdown("---") gr.Markdown( "**Data Source**: [GeoRisk Philippines API](https://hazardhunter.georisk.gov.ph/) | " "**Note**: Responses are cached for 1 hour to reduce API load." ) return demo if __name__ == "__main__": demo = create_interface() # Launch with MCP server enabled # MCP will automatically discover and expose get_disaster_risk() which returns structured JSON # Web UI uses _get_disaster_risk_html() internally (prefixed with _ so MCP ignores it) demo.launch(mcp_server=True)