GeoSight MCP Server

""" Geocoding utilities for resolving location names to coordinates. """ import httpx import structlog from typing import Optional from functools import lru_cache logger = structlog.get_logger(__name__) # OpenStreetMap Nominatim API NOMINATIM_URL = "https://nominatim.openstreetmap.org/search" NOMINATIM_REVERSE_URL = "https://nominatim.openstreetmap.org/reverse" async def get_location_info( location_name: Optional[str] = None, latitude: Optional[float] = None, longitude: Optional[float] = None, ) -> dict: """ Get geographic information about a location. Can geocode a location name to coordinates, or reverse geocode coordinates to location information. Args: location_name: Name of location to geocode latitude: Latitude for reverse geocoding longitude: Longitude for reverse geocoding Returns: Dictionary with location information """ async with httpx.AsyncClient() as client: if location_name and not (latitude and longitude): # Forward geocoding: name -> coordinates result = await geocode_location(client, location_name) elif latitude is not None and longitude is not None: # Reverse geocoding: coordinates -> name result = await reverse_geocode(client, latitude, longitude) else: return { "error": "Please provide either a location_name or both latitude and longitude", "summary": "❌ Invalid input: Need location name or coordinates", } return result async def geocode_location(client: httpx.AsyncClient, location_name: str) -> dict: """Convert location name to coordinates.""" try: response = await client.get( NOMINATIM_URL, params={ "q": location_name, "format": "json", "limit": 1, "addressdetails": 1, "extratags": 1, }, headers={"User-Agent": "GeoSight-MCP/1.0"}, timeout=10.0, ) response.raise_for_status() results = response.json() if not results: return { "error": f"Location not found: {location_name}", "summary": f"❌ Could not find location: {location_name}", } location = results[0] address = location.get("address", {}) lat = float(location["lat"]) lon = float(location["lon"]) return { "summary": f"📍 **{location['display_name']}**\n" f" Coordinates: {lat:.6f}, {lon:.6f}", "coordinates": { "latitude": lat, "longitude": lon, }, "display_name": location["display_name"], "type": location.get("type", "unknown"), "importance": location.get("importance", 0), "address": { "city": address.get("city") or address.get("town") or address.get("village"), "state": address.get("state"), "country": address.get("country"), "country_code": address.get("country_code", "").upper(), }, "bounding_box": { "south": float(location["boundingbox"][0]), "north": float(location["boundingbox"][1]), "west": float(location["boundingbox"][2]), "east": float(location["boundingbox"][3]), } if "boundingbox" in location else None, "statistics": { "latitude": lat, "longitude": lon, }, } except httpx.HTTPError as e: logger.error("geocoding_error", location=location_name, error=str(e)) return { "error": f"Geocoding failed: {str(e)}", "summary": f"❌ Failed to geocode location: {location_name}", } async def reverse_geocode( client: httpx.AsyncClient, latitude: float, longitude: float, ) -> dict: """Convert coordinates to location information.""" try: response = await client.get( NOMINATIM_REVERSE_URL, params={ "lat": latitude, "lon": longitude, "format": "json", "addressdetails": 1, "zoom": 10, }, headers={"User-Agent": "GeoSight-MCP/1.0"}, timeout=10.0, ) response.raise_for_status() location = response.json() if "error" in location: return { "error": location["error"], "summary": f"❌ No location found at coordinates: {latitude}, {longitude}", } address = location.get("address", {}) return { "summary": f"📍 **{location['display_name']}**\n" f" Coordinates: {latitude:.6f}, {longitude:.6f}", "coordinates": { "latitude": latitude, "longitude": longitude, }, "display_name": location["display_name"], "type": location.get("type", "unknown"), "address": { "city": address.get("city") or address.get("town") or address.get("village"), "state": address.get("state"), "country": address.get("country"), "country_code": address.get("country_code", "").upper(), }, "statistics": { "latitude": latitude, "longitude": longitude, }, } except httpx.HTTPError as e: logger.error("reverse_geocoding_error", lat=latitude, lon=longitude, error=str(e)) return { "error": f"Reverse geocoding failed: {str(e)}", "summary": f"❌ Failed to reverse geocode: {latitude}, {longitude}", } async def resolve_location( location_name: Optional[str] = None, latitude: Optional[float] = None, longitude: Optional[float] = None, ) -> tuple[float, float, str]: """ Resolve location to coordinates. Returns: Tuple of (latitude, longitude, display_name) Raises: ValueError if location cannot be resolved """ if latitude is not None and longitude is not None: # Coordinates provided directly return latitude, longitude, f"{latitude:.4f}, {longitude:.4f}" if location_name: # Geocode the location name result = await get_location_info(location_name=location_name) if "error" in result: raise ValueError(result["error"]) coords = result["coordinates"] return coords["latitude"], coords["longitude"], result["display_name"] raise ValueError("Either location_name or both latitude and longitude must be provided")