AEMET-MCP

Instructions

Implementation Reference

• src/aemet_mcp/server.py:216-247 (handler)

  The primary handler function decorated with @mcp.tool(), implementing the logic to find nearby weather stations using haversine distance calculation.

  @mcp.tool()
  async def find_nearby_stations(lat: float, lon: float, radio_km: float = 25):
      """
      Find weather stations within a given radius (in km) from a given geographic coordinate.
  
      Args:
          lat: Latitude in decimal degrees (e.g., 43.36)
          lon: Longitude in decimal degrees (e.g., -8.41)
          radio_km: Search radius in kilometers
      """
  
      url = f"{AEMET_API_BASE}/valores/climatologicos/inventarioestaciones/todasestaciones"
      estaciones = await make_aemet_request(url)
      if not estaciones:
          return {"error": "Could not retrieve station list."}
  
      resultado = []
      for est in estaciones:
          try:
              est_dict: dict = est  # type: ignore
              est_lat = sexagesimal_to_decimal(est_dict["latitud"])
              est_lon = sexagesimal_to_decimal(est_dict["longitud"])
              dist = haversine(lat, lon, est_lat, est_lon)
              if dist <= radio_km:
                  est_dict["distancia_km"] = round(dist, 2)
                  resultado.append(est_dict)
          except Exception:
              continue
  
      return resultado
      #return sorted(resultado, key=lambda x: x["distancia_km"])

• src/aemet_mcp/server.py:81-88 (helper)

  Supporting utility function to calculate the great-circle distance (haversine formula) between two latitude/longitude coordinates in kilometers. Used in find_nearby_stations.

  def haversine(lat1, lon1, lat2, lon2):
      # Distancia entre dos coordenadas en km
      R = 6371.0  # radio de la Tierra en km
      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

• src/aemet_mcp/server.py:90-109 (helper)

  Supporting utility function to parse AEMET's sexagesimal coordinate format (e.g., degrees minutes seconds with N/S/E/W) into decimal degrees. Used to process station coordinates.

  def sexagesimal_to_decimal(coord: str) -> float:
      """
      Convert AEMET-style sexagesimal string to decimal degrees.
      Examples:
          '424607N' -> 42.768611
          '070103W' -> -7.0175
      """
  
      direction = coord[-1]
      coord = coord[:-1]
  
      degrees = int(coord[:2])
      minutes = int(coord[2:4])
      seconds = int(coord[4:])
  
      decimal = degrees + minutes / 60 + seconds / 3600
      if direction in 'SW':
          decimal = -decimal
  
      return decimal

• src/aemet_mcp/server.py:55-77 (helper)

  General helper function to make authenticated asynchronous requests to the AEMET OpenData API, handling the two-step response process.

  async def make_aemet_request(url: str) -> dict[str, Any] | list[Any] | None:
      logger.info(f"make_aemet_request")
      headers = {
          "api_key": API_KEY,
          "Accept": "application/json"
      }
      async with httpx.AsyncClient() as client:
          try:
              response = await client.get(url, headers=headers, timeout=30.0)
              response.raise_for_status()
              data_info = response.json()
              if data_info.get("estado") == 200:
                  data_url = data_info.get("datos")
                  if data_url:
                      data_response = await client.get(data_url, timeout=30.0)
                      data_response.raise_for_status()
                      content = data_response.content.decode('latin1')
                      return json.loads(content)
              return None
          except Exception as e:
              logger.error(f"Error connecting to AEMET: {str(e)}")
              return None