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)
      }