Create app.py

app.py

@@ -0,0 +1,138 @@
+import asyncio
+import sqlite3
+import pandas as pd
+import gradio as gr
+from mcp.server import MCPServer
+from mcp.common.tool import Tool, ToolResult
+
+class EarthquakeDataTool(Tool):
+    """
+    A tool to retrieve earthquake data from the database with filtering capabilities.
+    """
+    def __init__(self, db_path):
+        super().__init__(
+            name="get_earthquake_data",
+            description="Fetches earthquake catalog data with optional filters.",
+            parameters={
+                "date": {"type": "string", "description": "Filter by date (YYYY-MM-DD)"},
+                "lat_min": {"type": "number", "description": "Minimum latitude"},
+                "lat_max": {"type": "number", "description": "Maximum latitude"},
+                "lon_min": {"type": "number", "description": "Minimum longitude"},
+                "lon_max": {"type": "number", "description": "Maximum longitude"},
+                "depth_min": {"type": "number", "description": "Minimum depth"},
+                "depth_max": {"type": "number", "description": "Maximum depth"},
+                "ML_min": {"type": "number", "description": "Minimum magnitude (ML)"},
+                "ML_max": {"type": "number", "description": "Maximum magnitude (ML)"},
+            },
+        )
+        self.db_path = db_path
+
+    async def __call__(self, **kwargs) -> ToolResult:
+        try:
+            conn = sqlite3.connect(self.db_path)
+            
+            query = "SELECT * FROM earthquakes WHERE 1=1"
+            params = []
+
+            # Handle filters
+            if kwargs.get("date"):
+                query += " AND date = ?"
+                params.append(kwargs["date"])
+            if kwargs.get("lat_min"):
+                query += " AND lat >= ?"
+                params.append(kwargs["lat_min"])
+            if kwargs.get("lat_max"):
+                query += " AND lat <= ?"
+                params.append(kwargs["lat_max"])
+            if kwargs.get("lon_min"):
+                query += " AND lon >= ?"
+                params.append(kwargs["lon_min"])
+            if kwargs.get("lon_max"):
+                query += " AND lon <= ?"
+                params.append(kwargs["lon_max"])
+            if kwargs.get("depth_min"):
+                query += " AND depth >= ?"
+                params.append(kwargs["depth_min"])
+            if kwargs.get("depth_max"):
+                query += " AND depth <= ?"
+                params.append(kwargs["depth_max"])
+            if kwargs.get("ML_min"):
+                query += " AND ML >= ?"
+                params.append(kwargs["ML_min"])
+            if kwargs.get("ML_max"):
+                query += " AND ML <= ?"
+                params.append(kwargs["ML_max"])
+
+            df = pd.read_sql_query(query, conn, params=tuple(params))
+            conn.close()
+
+            return ToolResult.success(df)
+        except Exception as e:
+            return ToolResult.error(str(e))
+
+class EarthquakeMCPServer(MCPServer):
+    def __init__(self, db_path):
+        super().__init__()
+        self.tool = EarthquakeDataTool(db_path)
+        self.add_tool(self.tool)
+
+# --- Gradio Interface ---
+def create_gradio_interface(server_tool):
+    async def fetch_data(date, lat_min, lat_max, lon_min, lon_max, depth_min, depth_max, ML_min, ML_max):
+        # Create a dictionary of filters, excluding empty values
+        filters = {
+            "date": date,
+            "lat_min": lat_min,
+            "lat_max": lat_max,
+            "lon_min": lon_min,
+            "lon_max": lon_max,
+            "depth_min": depth_min,
+            "depth_max": depth_max,
+            "ML_min": ML_min,
+            "ML_max": ML_max,
+        }
+        active_filters = {k: v for k, v in filters.items() if v is not None and v != ''}
+        
+        result = await server_tool(**active_filters)
+        if result.is_success:
+            return result.content
+        else:
+            return pd.DataFrame({"error": [result.content]})
+
+    with gr.Blocks() as iface:
+        gr.Markdown("# Earthquake Data Explorer")
+        gr.Markdown("Use the filters below to search the earthquake catalog.")
+        
+        with gr.Row():
+            date = gr.Textbox(label="Date (YYYY-MM-DD)")
+        with gr.Row():
+            lat_min = gr.Number(label="Min Latitude")
+            lat_max = gr.Number(label="Max Latitude")
+        with gr.Row():
+            lon_min = gr.Number(label="Min Longitude")
+            lon_max = gr.Number(label="Max Longitude")
+        with gr.Row():
+            depth_min = gr.Number(label="Min Depth")
+            depth_max = gr.Number(label="Max Depth")
+        with gr.Row():
+            ML_min = gr.Number(label="Min Magnitude (ML)")
+            ML_max = gr.Number(label="Max Magnitude (ML)")
+        
+        filter_button = gr.Button("Filter Data")
+        
+        output_df = gr.DataFrame(label="Filtered Results")
+        
+        filter_button.click(
+            fn=fetch_data,
+            inputs=[date, lat_min, lat_max, lon_min, lon_max, depth_min, depth_max, ML_min, ML_max],
+            outputs=output_df
+        )
+    return iface
+
+# --- Main execution ---
+if __name__ == "__main__":
+    # In a real-world scenario, the server might run as a separate process.
+    # For this Gradio app, we'll instantiate the tool directly.
+    server = EarthquakeMCPServer('earthquake_data.db')
+    gradio_app = create_gradio_interface(server.tool)
+    gradio_app.launch(server_name="0.0.0.0")