"""
GeoSight MCP Server - Satellite Imagery Analysis
"""
import asyncio
import json
from typing import Optional
from mcp.server import Server
from mcp.server.stdio import stdio_server
from mcp.types import Tool, TextContent
from geosight.config import settings
from geosight.utils.logging import setup_logging, get_logger
logger = get_logger(__name__)
# Create MCP server
server = Server("geosight")
# Tool definitions
TOOLS = [
Tool(
name="search_imagery",
description="Search for satellite imagery",
inputSchema={
"type": "object",
"properties": {
"location_name": {"type": "string"},
"latitude": {"type": "number"},
"longitude": {"type": "number"},
"start_date": {"type": "string"},
"end_date": {"type": "string"},
},
"required": ["start_date", "end_date"]
}
),
Tool(
name="calculate_ndvi",
description="Calculate vegetation index (NDVI)",
inputSchema={
"type": "object",
"properties": {
"location_name": {"type": "string"},
"latitude": {"type": "number"},
"longitude": {"type": "number"},
"start_date": {"type": "string"},
"end_date": {"type": "string"},
"radius_km": {"type": "number", "default": 10}
},
"required": ["start_date", "end_date"]
}
),
Tool(
name="calculate_ndwi",
description="Calculate water index (NDWI)",
inputSchema={
"type": "object",
"properties": {
"location_name": {"type": "string"},
"latitude": {"type": "number"},
"longitude": {"type": "number"},
"start_date": {"type": "string"},
"end_date": {"type": "string"},
},
"required": ["start_date", "end_date"]
}
),
Tool(
name="detect_land_cover",
description="Classify land cover types using ML",
inputSchema={
"type": "object",
"properties": {
"location_name": {"type": "string"},
"latitude": {"type": "number"},
"longitude": {"type": "number"},
"date": {"type": "string"},
"radius_km": {"type": "number", "default": 10}
},
"required": ["date"]
}
),
Tool(
name="detect_changes",
description="Detect changes between two dates",
inputSchema={
"type": "object",
"properties": {
"location_name": {"type": "string"},
"latitude": {"type": "number"},
"longitude": {"type": "number"},
"date_before": {"type": "string"},
"date_after": {"type": "string"},
},
"required": ["date_before", "date_after"]
}
),
Tool(
name="get_location_info",
description="Get coordinates for a location name",
inputSchema={
"type": "object",
"properties": {
"location_name": {"type": "string"},
"latitude": {"type": "number"},
"longitude": {"type": "number"},
}
}
),
]
@server.list_tools()
async def list_tools():
return TOOLS
@server.call_tool()
async def call_tool(name: str, arguments: dict):
logger.info(f"Tool called: {name}", arguments=arguments)
try:
if name == "search_imagery":
from geosight.tools.imagery import search_imagery
result = await search_imagery(**arguments)
elif name == "calculate_ndvi":
from geosight.tools.indices import calculate_ndvi
result = await calculate_ndvi(**arguments)
elif name == "calculate_ndwi":
from geosight.tools.indices import calculate_ndwi
result = await calculate_ndwi(**arguments)
elif name == "detect_land_cover":
from geosight.tools.classification import detect_land_cover
result = await detect_land_cover(**arguments)
elif name == "detect_changes":
from geosight.tools.change_detection import detect_changes
result = await detect_changes(**arguments)
elif name == "get_location_info":
from geosight.tools.geocoding import get_location_info
result = await get_location_info(**arguments)
else:
result = {"error": f"Unknown tool: {name}"}
return [TextContent(
type="text",
text=result.get("summary", json.dumps(result, indent=2))
)]
except Exception as e:
logger.error(f"Tool error: {name}", error=str(e))
return [TextContent(type="text", text=f"Error: {str(e)}")]
async def run_stdio():
"""Run in stdio mode for Claude Desktop."""
async with stdio_server() as (read_stream, write_stream):
await server.run(read_stream, write_stream, server.create_initialization_options())
def run_http():
"""Run in HTTP mode for cloud deployment."""
from http.server import HTTPServer, BaseHTTPRequestHandler
import json
class HealthHandler(BaseHTTPRequestHandler):
def do_GET(self):
if self.path == "/health":
self.send_response(200)
self.send_header("Content-Type", "application/json")
self.end_headers()
self.wfile.write(json.dumps({"status": "healthy", "service": "geosight"}).encode())
elif self.path == "/":
self.send_response(200)
self.send_header("Content-Type", "application/json")
self.end_headers()
response = {
"service": "GeoSight MCP Server",
"version": "1.0.0",
"status": "running",
"tools": [t.name for t in TOOLS],
"docs": "https://github.com/your-repo/geosight-mcp"
}
self.wfile.write(json.dumps(response, indent=2).encode())
else:
self.send_response(404)
self.end_headers()
def log_message(self, format, *args):
logger.info(f"HTTP: {args[0]}")
port = settings.server.port
httpd = HTTPServer(("0.0.0.0", port), HealthHandler)
logger.info(f"HTTP server running on port {port}")
httpd.serve_forever()
def main():
setup_logging(level=settings.log_level)
logger.info(
"starting_server",
environment=settings.environment,
mode=settings.server.mode
)
try:
if settings.server.mode == "stdio":
asyncio.run(run_stdio())
elif settings.server.mode == "http":
run_http()
else:
raise ValueError(f"Unknown server mode: {settings.server.mode}")
except Exception as e:
logger.error("server_crash", error=str(e))
raise
if __name__ == "__main__":
main()
