JMA Data MCP

Instructions

Implementation Reference

• jma_data_mcp/server.py:219-265 (handler)

  MCP tool handler and registration for get_historical_weather. Handles input validation (datetime parsing), calls core fetch function, adds station metadata, and returns structured data.
  @mcp.tool() async def get_historical_weather( station_code: str, target_datetime: str, ) -> dict: """Get historical weather data for a specific date and time. Data is available for approximately the past 1-2 weeks. Args: station_code: Station code (e.g., '44132' for Tokyo) target_datetime: Target datetime in ISO format (e.g., '2025-12-01T12:00:00') or 'YYYY-MM-DD HH:MM' format. Time is in JST. Returns: Weather data for the specified time """ try: if "T" in target_datetime: target_time = datetime.fromisoformat(target_datetime.replace("Z", "+00:00")) else: for fmt in ["%Y-%m-%d %H:%M", "%Y-%m-%d %H:%M:%S", "%Y/%m/%d %H:%M"]: try: target_time = datetime.strptime(target_datetime, fmt) break except ValueError: continue else: raise ValueError(f"Could not parse datetime: {target_datetime}") if target_time.tzinfo is None: target_time = target_time.replace(tzinfo=JST) except Exception as e: return { "error": f"Invalid datetime format: {e}", "hint": "Use ISO format (e.g., '2025-12-01T12:00:00') or 'YYYY-MM-DD HH:MM'", } historical_data = await fetch_historical_amedas_data(station_code, target_time) station_info = get_station(station_code) if station_info: historical_data["station_info"] = station_info return historical_data
• jma_data_mcp/weather.py:220-264 (helper)

  Core implementation fetching historical weather data from JMA AMeDAS API. Rounds time to 10-min intervals, makes HTTP request to historical endpoint, handles missing data, and parses response using _parse_station_data.
  async def fetch_historical_amedas_data( station_code: str, target_time: datetime ) -> dict[str, Any]: """ Fetch historical AMeDAS data for a specific time. Args: station_code: Station code (e.g., '44132' for Tokyo) target_time: Target datetime (available for approximately past 1-2 weeks) Returns: Dictionary with observation data for the specified time """ # Round to 10-minute intervals target_time = target_time.replace( minute=(target_time.minute // 10) * 10, second=0, microsecond=0 ) time_str = format_time_for_api(target_time) url = f'https://www.jma.go.jp/bosai/amedas/data/map/{time_str}.json' async with httpx.AsyncClient() as client: response = await client.get(url, timeout=30.0) response.raise_for_status() raw_data = response.json() station_data = raw_data.get(station_code) if station_data is None: return { "error": f"No data found for station {station_code} at {target_time.isoformat()}", "observation_time": target_time.isoformat() } result = { "observation_time": target_time.isoformat(), "observation_time_jst": target_time.strftime('%Y-%m-%d %H:%M JST'), "station_code": station_code, "data": _parse_station_data(station_code, station_data) } return result
• jma_data_mcp/weather.py:343-436 (helper)

  Supporting helper that parses raw JSON data from JMA API into a structured dictionary with weather elements like temperature, wind, precipitation, snow, etc., including units and wind directions.
  def _parse_station_data(code: str, data: dict[str, Any]) -> dict[str, Any]: """Parse raw station data into structured format.""" station_data: dict[str, Any] = {"code": code} # Temperature (℃) if "temp" in data: station_data["temperature"] = { "value": parse_observation_value(data["temp"]), "unit": "℃" } # Humidity (%) if "humidity" in data: station_data["humidity"] = { "value": parse_observation_value(data["humidity"]), "unit": "%" } # Pressure (hPa) if "pressure" in data: station_data["pressure"] = { "value": parse_observation_value(data["pressure"]), "unit": "hPa" } # Sea level pressure (hPa) if "normalPressure" in data: station_data["sea_level_pressure"] = { "value": parse_observation_value(data["normalPressure"]), "unit": "hPa" } # Wind if "wind" in data: wind_speed = parse_observation_value(data["wind"]) wind_dir_code = parse_observation_value(data.get("windDirection", [None, None])) wind_dir = None wind_dir_ja = None if wind_dir_code is not None: wind_dir_code = int(wind_dir_code) wind_dir = WIND_DIRECTIONS.get(wind_dir_code) wind_dir_ja = WIND_DIRECTIONS_JA.get(wind_dir_code) station_data["wind"] = { "speed": wind_speed, "speed_unit": "m/s", "direction": wind_dir, "direction_ja": wind_dir_ja, "direction_code": wind_dir_code } # Precipitation precipitation = {} if "precipitation10m" in data: precipitation["10min"] = parse_observation_value(data["precipitation10m"]) if "precipitation1h" in data: precipitation["1h"] = parse_observation_value(data["precipitation1h"]) if "precipitation3h" in data: precipitation["3h"] = parse_observation_value(data["precipitation3h"]) if "precipitation24h" in data: precipitation["24h"] = parse_observation_value(data["precipitation24h"]) if precipitation: station_data["precipitation"] = { **precipitation, "unit": "mm" } # Sunshine if "sun1h" in data: station_data["sunshine"] = { "1h": parse_observation_value(data["sun1h"]), "unit": "hours" } # Snow snow = {} if "snow" in data: snow["depth"] = parse_observation_value(data["snow"]) if "snow1h" in data: snow["1h"] = parse_observation_value(data["snow1h"]) if "snow6h" in data: snow["6h"] = parse_observation_value(data["snow6h"]) if "snow12h" in data: snow["12h"] = parse_observation_value(data["snow12h"]) if "snow24h" in data: snow["24h"] = parse_observation_value(data["snow24h"]) if snow: station_data["snow"] = { **snow, "unit": "cm" } return station_data