GeoSight MCP Server

""" GeoSight MCP Server - Main Entry Point This module implements the MCP server that exposes satellite imagery analysis tools through the Model Context Protocol. """ import asyncio import logging import sys from contextlib import asynccontextmanager from typing import Any import structlog from mcp.server import Server from mcp.server.stdio import stdio_server from mcp.types import ( CallToolResult, ListToolsResult, TextContent, ImageContent, Tool, ) from geosight.config import settings from geosight.tools import ( search_imagery, calculate_ndvi, calculate_ndwi, calculate_ndbi, detect_land_cover, detect_changes, detect_objects, generate_report, get_location_info, ) from geosight.utils.logging import setup_logging # Setup structured logging setup_logging(settings.log_level) logger = structlog.get_logger(__name__) def create_server() -> Server: """Create and configure the MCP server instance.""" server = Server("geosight") @server.list_tools() async def list_tools() -> ListToolsResult: """List all available satellite imagery analysis tools.""" tools = [ Tool( name="search_imagery", description="""Search for available satellite imagery for a given location and date range. Use this tool to find what imagery is available before running analysis. Supports Sentinel-2, Landsat 8/9, and Sentinel-1 (radar) data. Example queries: - "Find imagery for New Delhi from last month" - "Search for cloud-free Sentinel-2 images of Mumbai" """, inputSchema={ "type": "object", "properties": { "latitude": { "type": "number", "description": "Latitude of the center point (-90 to 90)", "minimum": -90, "maximum": 90, }, "longitude": { "type": "number", "description": "Longitude of the center point (-180 to 180)", "minimum": -180, "maximum": 180, }, "location_name": { "type": "string", "description": "Name of the location (e.g., 'Mumbai, India'). Will be geocoded if lat/lon not provided.", }, "start_date": { "type": "string", "description": "Start date in YYYY-MM-DD format", "pattern": "^\\d{4}-\\d{2}-\\d{2}$", }, "end_date": { "type": "string", "description": "End date in YYYY-MM-DD format", "pattern": "^\\d{4}-\\d{2}-\\d{2}$", }, "max_cloud_cover": { "type": "number", "description": "Maximum cloud cover percentage (0-100)", "minimum": 0, "maximum": 100, "default": 20, }, "data_source": { "type": "string", "enum": ["sentinel-2", "landsat-8", "landsat-9", "sentinel-1"], "description": "Satellite data source", "default": "sentinel-2", }, }, "required": ["start_date", "end_date"], }, ), Tool( name="calculate_ndvi", description="""Calculate NDVI (Normalized Difference Vegetation Index) for vegetation health analysis. NDVI values range from -1 to 1: - 0.6 to 1.0: Dense, healthy vegetation - 0.3 to 0.6: Moderate vegetation - 0.1 to 0.3: Sparse vegetation - -0.1 to 0.1: Bare soil, rocks, sand - -1 to -0.1: Water, snow, clouds Use cases: - Agricultural monitoring and crop health - Deforestation tracking - Drought assessment - Urban green space analysis """, inputSchema={ "type": "object", "properties": { "latitude": {"type": "number", "minimum": -90, "maximum": 90}, "longitude": {"type": "number", "minimum": -180, "maximum": 180}, "location_name": {"type": "string"}, "radius_km": { "type": "number", "description": "Analysis radius in kilometers", "minimum": 1, "maximum": 100, "default": 10, }, "start_date": {"type": "string", "pattern": "^\\d{4}-\\d{2}-\\d{2}$"}, "end_date": {"type": "string", "pattern": "^\\d{4}-\\d{2}-\\d{2}$"}, "resolution": { "type": "integer", "description": "Output resolution in meters", "enum": [10, 20, 60], "default": 10, }, }, "required": ["start_date", "end_date"], }, ), Tool( name="calculate_ndwi", description="""Calculate NDWI (Normalized Difference Water Index) for water body detection. NDWI values: - > 0.3: Open water - 0.0 to 0.3: Flooding, wetlands - < 0.0: Non-water surfaces Use cases: - Flood mapping and monitoring - Water body extent tracking - Wetland mapping - Coastal change analysis """, inputSchema={ "type": "object", "properties": { "latitude": {"type": "number", "minimum": -90, "maximum": 90}, "longitude": {"type": "number", "minimum": -180, "maximum": 180}, "location_name": {"type": "string"}, "radius_km": {"type": "number", "minimum": 1, "maximum": 100, "default": 10}, "start_date": {"type": "string", "pattern": "^\\d{4}-\\d{2}-\\d{2}$"}, "end_date": {"type": "string", "pattern": "^\\d{4}-\\d{2}-\\d{2}$"}, "water_threshold": { "type": "number", "description": "Threshold for water classification", "default": 0.3, }, }, "required": ["start_date", "end_date"], }, ), Tool( name="calculate_ndbi", description="""Calculate NDBI (Normalized Difference Built-up Index) for urban area detection. NDBI helps identify built-up/urban areas: - High values: Dense urban areas, roads, buildings - Low values: Vegetation, water, bare soil Use cases: - Urban expansion tracking - Infrastructure development monitoring - Land use change analysis """, inputSchema={ "type": "object", "properties": { "latitude": {"type": "number", "minimum": -90, "maximum": 90}, "longitude": {"type": "number", "minimum": -180, "maximum": 180}, "location_name": {"type": "string"}, "radius_km": {"type": "number", "minimum": 1, "maximum": 100, "default": 10}, "start_date": {"type": "string", "pattern": "^\\d{4}-\\d{2}-\\d{2}$"}, "end_date": {"type": "string", "pattern": "^\\d{4}-\\d{2}-\\d{2}$"}, }, "required": ["start_date", "end_date"], }, ), Tool( name="detect_land_cover", description="""Classify land cover into categories using machine learning. Categories detected: - Forest (deciduous, evergreen, mixed) - Water (rivers, lakes, reservoirs) - Urban (residential, commercial, industrial) - Agriculture (cropland, pasture) - Barren (desert, rock, sand) - Wetlands - Snow/Ice Returns a classification map with area statistics for each category. """, inputSchema={ "type": "object", "properties": { "latitude": {"type": "number", "minimum": -90, "maximum": 90}, "longitude": {"type": "number", "minimum": -180, "maximum": 180}, "location_name": {"type": "string"}, "radius_km": {"type": "number", "minimum": 1, "maximum": 50, "default": 10}, "date": { "type": "string", "description": "Target date for classification", "pattern": "^\\d{4}-\\d{2}-\\d{2}$", }, "model": { "type": "string", "enum": ["eurosat", "deepglobe", "custom"], "default": "eurosat", }, }, "required": ["date"], }, ), Tool( name="detect_changes", description="""Detect changes between two time periods using bi-temporal analysis. Identifies: - Deforestation / Reforestation - Urban expansion / Construction - Flood events - Agricultural changes - Mining / Excavation Returns a change map highlighting areas of significant change with statistics. """, inputSchema={ "type": "object", "properties": { "latitude": {"type": "number", "minimum": -90, "maximum": 90}, "longitude": {"type": "number", "minimum": -180, "maximum": 180}, "location_name": {"type": "string"}, "radius_km": {"type": "number", "minimum": 1, "maximum": 50, "default": 10}, "date_before": { "type": "string", "description": "Earlier date for comparison", "pattern": "^\\d{4}-\\d{2}-\\d{2}$", }, "date_after": { "type": "string", "description": "Later date for comparison", "pattern": "^\\d{4}-\\d{2}-\\d{2}$", }, "change_type": { "type": "string", "enum": ["all", "vegetation", "urban", "water"], "default": "all", }, "sensitivity": { "type": "string", "enum": ["low", "medium", "high"], "default": "medium", }, }, "required": ["date_before", "date_after"], }, ), Tool( name="detect_objects", description="""Detect specific objects in satellite imagery using deep learning. Detectable objects: - Ships and vessels - Aircraft - Vehicles - Buildings and structures - Solar panels / Solar farms - Storage tanks - Sports fields Returns bounding boxes with confidence scores for detected objects. """, inputSchema={ "type": "object", "properties": { "latitude": {"type": "number", "minimum": -90, "maximum": 90}, "longitude": {"type": "number", "minimum": -180, "maximum": 180}, "location_name": {"type": "string"}, "radius_km": {"type": "number", "minimum": 0.5, "maximum": 20, "default": 5}, "date": {"type": "string", "pattern": "^\\d{4}-\\d{2}-\\d{2}$"}, "object_types": { "type": "array", "items": { "type": "string", "enum": [ "ship", "aircraft", "vehicle", "building", "solar_panel", "storage_tank", "sports_field", ], }, "description": "Types of objects to detect", "default": ["ship", "aircraft", "building"], }, "confidence_threshold": { "type": "number", "description": "Minimum confidence score (0-1)", "minimum": 0, "maximum": 1, "default": 0.5, }, }, "required": ["date"], }, ), Tool( name="generate_report", description="""Generate a comprehensive analysis report with maps and statistics. Report includes: - Location overview and map - Satellite imagery preview - Analysis results (NDVI, land cover, etc.) - Change detection if applicable - Statistical summaries - Recommendations Output formats: PDF, HTML, or Markdown """, inputSchema={ "type": "object", "properties": { "latitude": {"type": "number", "minimum": -90, "maximum": 90}, "longitude": {"type": "number", "minimum": -180, "maximum": 180}, "location_name": {"type": "string"}, "radius_km": {"type": "number", "minimum": 1, "maximum": 50, "default": 10}, "start_date": {"type": "string", "pattern": "^\\d{4}-\\d{2}-\\d{2}$"}, "end_date": {"type": "string", "pattern": "^\\d{4}-\\d{2}-\\d{2}$"}, "analyses": { "type": "array", "items": { "type": "string", "enum": ["ndvi", "ndwi", "land_cover", "change_detection"], }, "description": "Analyses to include in report", "default": ["ndvi", "land_cover"], }, "output_format": { "type": "string", "enum": ["pdf", "html", "markdown"], "default": "pdf", }, "report_title": { "type": "string", "description": "Custom title for the report", }, }, "required": ["start_date", "end_date"], }, ), Tool( name="get_location_info", description="""Get geographic information about a location. Geocodes location names to coordinates and provides: - Coordinates (latitude, longitude) - Administrative boundaries - Elevation data - Climate zone - Time zone Use this to convert place names to coordinates for other tools. """, inputSchema={ "type": "object", "properties": { "location_name": { "type": "string", "description": "Name of the location to geocode (e.g., 'Mumbai, India')", }, "latitude": {"type": "number", "minimum": -90, "maximum": 90}, "longitude": {"type": "number", "minimum": -180, "maximum": 180}, }, }, ), ] return ListToolsResult(tools=tools) @server.call_tool() async def call_tool(name: str, arguments: dict[str, Any]) -> CallToolResult: """Execute a satellite imagery analysis tool.""" logger.info("tool_called", tool=name, arguments=arguments) try: # Route to appropriate tool handler tool_handlers = { "search_imagery": search_imagery, "calculate_ndvi": calculate_ndvi, "calculate_ndwi": calculate_ndwi, "calculate_ndbi": calculate_ndbi, "detect_land_cover": detect_land_cover, "detect_changes": detect_changes, "detect_objects": detect_objects, "generate_report": generate_report, "get_location_info": get_location_info, } if name not in tool_handlers: return CallToolResult( content=[TextContent(type="text", text=f"Unknown tool: {name}")], isError=True, ) handler = tool_handlers[name] result = await handler(**arguments) # Format response based on result type content = [] if isinstance(result, dict): # Add text summary if "summary" in result: content.append(TextContent(type="text", text=result["summary"])) # Add statistics as formatted text if "statistics" in result: stats_text = format_statistics(result["statistics"]) content.append(TextContent(type="text", text=stats_text)) # Add image if available if "image_base64" in result: content.append( ImageContent( type="image", data=result["image_base64"], mimeType=result.get("image_mime_type", "image/png"), ) ) # Add map URL if available if "map_url" in result: content.append( TextContent( type="text", text=f"\n📍 Interactive Map: {result['map_url']}" ) ) # Add any additional text content if "details" in result: content.append(TextContent(type="text", text=result["details"])) else: content.append(TextContent(type="text", text=str(result))) logger.info("tool_completed", tool=name, success=True) return CallToolResult(content=content) except Exception as e: logger.error("tool_error", tool=name, error=str(e), exc_info=True) return CallToolResult( content=[TextContent(type="text", text=f"Error executing {name}: {str(e)}")], isError=True, ) return server def format_statistics(stats: dict[str, Any]) -> str: """Format statistics dictionary as readable text.""" lines = ["\n📊 **Statistics:**"] for key, value in stats.items(): # Format key formatted_key = key.replace("_", " ").title() # Format value if isinstance(value, float): if "percent" in key.lower() or "pct" in key.lower(): formatted_value = f"{value:.1f}%" else: formatted_value = f"{value:.4f}" elif isinstance(value, dict): formatted_value = ", ".join(f"{k}: {v}" for k, v in value.items()) else: formatted_value = str(value) lines.append(f" • {formatted_key}: {formatted_value}") return "\n".join(lines) async def run_server(): """Run the MCP server.""" server = create_server() logger.info( "starting_server", mode=settings.server.mode, environment=settings.environment, ) if settings.server.mode == "stdio": async with stdio_server() as (read_stream, write_stream): await server.run( read_stream, write_stream, server.create_initialization_options(), ) else: # HTTP/SSE mode would be implemented here raise NotImplementedError(f"Server mode {settings.server.mode} not yet implemented") def main(): """Main entry point.""" try: asyncio.run(run_server()) except KeyboardInterrupt: logger.info("server_shutdown", reason="keyboard_interrupt") except Exception as e: logger.error("server_crash", error=str(e), exc_info=True) sys.exit(1) if __name__ == "__main__": main()