JMA Data MCP

Instructions

Implementation Reference

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

  The MCP tool handler for get_historical_weather. Parses the target_datetime, fetches historical data using fetch_historical_amedas_data, adds station information, and returns the result. Also serves as registration via @mcp.tool() decorator.

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

  Helper function that performs the core logic: rounds time to 10-min interval, fetches JSON data from JMA AMeDAS API, parses the station data using _parse_station_data, and formats the response.

  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