find_nearby_stations

Locate nearby weather stations within a specified radius from a given geographic coordinate using latitude, longitude, and distance in kilometers.

Instructions

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

Input Schema

TableJSON Schema

Name	Required	Description	Default
`lat`	Yes
`lon`	Yes
`radio_km`	No

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

AEMET-MCP