OpenStreetMap MCP Server

Instructions

Implementation Reference

• src/osm_mcp_server/server.py:1054-1224 (handler)

  The handler function decorated with @mcp.tool() that implements the find_ev_charging_stations tool. It queries the Overpass API for OSM charging stations within a bounding box derived from the center coordinates and radius, processes and filters the results based on connector types and minimum power, calculates haversine distances, and returns a structured dictionary with station details.
  async def find_ev_charging_stations( latitude: float, longitude: float, ctx: Context, radius: float = 5000, connector_types: List[str] = None, min_power: float = None ) -> Dict[str, Any]: """ Locate electric vehicle charging stations near a specific location. This specialized search tool identifies EV charging infrastructure within a specified distance from a location. Results can be filtered by connector type (Tesla, CCS, CHAdeMO, etc.) and minimum power delivery. Essential for EV owners planning trips or evaluating potential charging stops. Args: latitude: Center point latitude (decimal degrees) longitude: Center point longitude (decimal degrees) radius: Search radius in meters (defaults to 5000m/5km) connector_types: Optional list of specific connector types to filter by (e.g., ["type2", "ccs", "tesla"]) min_power: Minimum charging power in kW Returns: List of charging stations with: - Location name and operator - Available connector types - Charging speeds - Number of charging points - Access restrictions - Other relevant metadata """ osm_client = ctx.request_context.lifespan_context.osm_client # Convert radius to bounding box lat_delta = radius / 111000 lon_delta = radius / (111000 * math.cos(math.radians(latitude))) bbox = ( longitude - lon_delta, latitude - lat_delta, longitude + lon_delta, latitude + lat_delta ) # Build Overpass query for EV charging stations overpass_url = "https://overpass-api.de/api/interpreter" query = f""" [out:json]; ( node["amenity"="charging_station"]({{bbox}}); way["amenity"="charging_station"]({{bbox}}); ); out body; """ query = query.replace("{bbox}", f"{bbox[1]},{bbox[0]},{bbox[3]},{bbox[2]}") async with aiohttp.ClientSession() as session: async with session.post(overpass_url, data={"data": query}) as response: if response.status == 200: data = await response.json() stations = data.get("elements", []) else: raise Exception(f"Failed to find charging stations: {response.status}") # Process and filter results results = [] for station in stations: tags = station.get("tags", {}) # Get coordinates based on feature type coords = {} if station.get("type") == "node": coords = { "latitude": station.get("lat"), "longitude": station.get("lon") } elif "center" in station: coords = { "latitude": station.get("center", {}).get("lat"), "longitude": station.get("center", {}).get("lon") } # Skip if no valid coordinates if not coords: continue # Extract connector information connectors = [] for key, value in tags.items(): if key.startswith("socket:"): connector_type = key.split(":", 1)[1] connectors.append({ "type": connector_type, "count": value if value.isdigit() else 1 }) # Filter by connector type if specified if connector_types: has_matching_connector = False for connector in connectors: if connector["type"] in connector_types: has_matching_connector = True break if not has_matching_connector: continue # Extract power information power = None if "maxpower" in tags: try: power = float(tags["maxpower"]) except ValueError: pass # Filter by minimum power if specified if min_power is not None and (power is None or power < min_power): continue # Calculate distance from search point from math import radians, sin, cos, sqrt, asin def haversine(lat1, lon1, lat2, lon2): R = 6371000 # Earth radius in meters dLat = radians(lat2 - lat1) dLon = radians(lon2 - lon1) a = sin(dLat/2)**2 + cos(radians(lat1)) * cos(radians(lat2)) * sin(dLon/2)**2 c = 2 * asin(sqrt(a)) return R * c distance = haversine(latitude, longitude, coords["latitude"], coords["longitude"]) results.append({ "id": station.get("id"), "name": tags.get("name", "Unnamed Charging Station"), "operator": tags.get("operator", "Unknown"), "coordinates": coords, "distance": round(distance, 1), "connectors": connectors, "capacity": tags.get("capacity", "Unknown"), "power": power, "fee": tags.get("fee", "Unknown"), "access": tags.get("access", "public"), "opening_hours": tags.get("opening_hours", "Unknown"), "address": { "street": tags.get("addr:street", ""), "housenumber": tags.get("addr:housenumber", ""), "city": tags.get("addr:city", ""), "postcode": tags.get("addr:postcode", "") }, "tags": tags }) # Sort by distance results.sort(key=lambda x: x["distance"]) return { "query": { "latitude": latitude, "longitude": longitude, "radius": radius, "connector_types": connector_types, "min_power": min_power }, "stations": results, "count": len(results) }