Weather MCP MY

Instructions

Args:
    datetime_start: The earliest timestamp in the form of <YYYY-MM-DD HH:mm:ss> (inclusive) from which to retrieve weather warnings. If omitted, defaults to the current date.
    datetime_end: The latest timestamp in the form of <YYYY-MM-DD HH:mm:ss> (inclusive) to stop retrieving the weather warnings. If omitted, defaults to the current date.

Implementation Reference

• weather.py:46-46 (registration)

  Registers the 'get_warning' tool using the @mcp.tool() decorator.

  @mcp.tool()

• weather.py:47-79 (handler)

  The handler function that implements the 'get_warning' tool logic: validates input datetimes, fetches data from the weather warning API, formats the warnings, and returns them as a joined string.

  async def get_warning(datetime_start: str = None, datetime_end: str = None) -> str:
      """Retrieve general weather warnings issued within a specified date range.
  
      Args:
          datetime_start: The earliest timestamp in the form of <YYYY-MM-DD HH:mm:ss> (inclusive) from which to retrieve weather warnings. If omitted, defaults to the current date.
          datetime_end: The latest timestamp in the form of <YYYY-MM-DD HH:mm:ss> (inclusive) to stop retrieving the weather warnings. If omitted, defaults to the current date.
      """
      if not datetime_start:
          datetime_start = current_date() + " 00:00:01"
      elif not validate_datetime(datetime_start):
          return "Wrong `datetime_start` format given. Accepted format is 'YYYY-MM-DD HH:mm:ss'."
  
      if not datetime_end:
          datetime_end = current_date() + " 23:59:59"
      elif not validate_datetime(datetime_end):
          return "Wrong `datetime_end` format given. Accepted format is 'YYYY-MM-DD HH:mm:ss'."
  
      warning_url = f"{GOV_API_BASE}/weather/warning"
      warning_data = await make_api_request(warning_url, {
                                              "meta": "true",
                                              "sort": "-warning_issue__issued",
                                              "timestamp_start": f"{datetime_start}@warning_issue__issued",
                                              "timestamp_end": f"{datetime_end}@warning_issue__issued",
                                          })
  
      if not warning_data or "data" not in warning_data:
          return "Unable to fetch warning data for this time period."
  
      if not warning_data["data"]:
          return "No active warning data for this time period."
  
      warnings = [format_warning(warn) for warn in warning_data["data"]]
      return "\n---\n".join(warnings)

• helpers.py:20-31 (helper)

  Helper function to format individual weather warning data into a human-readable string, used within the get_warning handler.

  def format_warning(warning_res: dict) -> str:
      """Format a warning json response into a readable string."""
      warning_issue = warning_res["warning_issue"]
      return f"""
  Warning Issue Date: {warning_issue.get('issued', 'Unknown')}
  Title: {warning_issue.get('title_en', 'Unknown')}
  Is Valid From: {warning_res.get('valid_from', 'Unknown')}
  Is Valid To: {warning_res.get('valid_to', 'Unknown')}
  Heading: {warning_res.get('heading_en', 'Unknown')}
  Details: {warning_res.get('text_en', 'Unknown')}
  Instruction: {warning_res.get('instruction_en', 'Unknown')}
  """

• helpers.py:76-86 (helper)

  Helper function to make asynchronous HTTP requests to the API, used by get_warning to fetch warning data.

  async def make_api_request(url: str, params: dict[str, str]) -> dict[str, Any] | None:
      """Make a request to the Malaysia Government API with proper error handling."""
      async with httpx.AsyncClient() as client:
          try:
              response = await client.get(url, params=params, timeout=30.0, follow_redirects=True)
              print(response)
              response.raise_for_status()
              return response.json()
          except Exception as ex:
              return None

• helpers.py:95-101 (helper)

  Helper function to validate datetime format, used in get_warning for input validation.

  def validate_datetime(datetime_text: str) -> bool:
      """Validate date string format."""
      try:
          datetime.strptime(datetime_text, "%Y-%m-%d %H:%M:%S")
          return True
      except ValueError:
          return False