Weather MCP MY

Instructions

Implementation Reference

• weather.py:118-153 (handler)

  The handler function decorated with @mcp.tool() that implements the get_earthquake_news tool. It validates input dates, makes an API request to the Malaysian government weather API for earthquake warnings, and formats the results using helper functions.
  @mcp.tool() async def get_earthquake_news(location: str, datetime_start: str = None, datetime_end: str = None) -> str: """Fetch earthquake news for a given location within a specified date range. Args: location: Name or identifier of the place where the earthquake(s) occurred. date_start: The earliest timestamp in the form of <YYYY-MM-DD HH:mm:ss> (inclusive) to start searching for earthquake news. If omitted, defaults to the current date. date_end: The latest timestamp in the form of <YYYY-MM-DD HH:mm:ss> (inclusive) to stop searching for earthquake news. 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'." earthquake_url = f"{GOV_API_BASE}/weather/warning/earthquake" data = await make_api_request(earthquake_url, { "meta": "true", "sort": "-utcdatetime", "timestamp_start": f"{datetime_start}@utcdatetime", "timestamp_end": f"{datetime_end}@utcdatetime", "icontains": f"{location}@location" }) if not data or "data" not in data: return "Unable to fetch earthquake news or no earthquake news found." if not data["data"]: return "No active earthquake news for this location." earthquakes = [format_earthquake_data(earthquake_desc) for earthquake_desc in data["data"]] return "\n---\n".join(earthquakes)
• helpers.py:33-47 (helper)

  Helper function used by the get_earthquake_news handler to format individual earthquake data records into human-readable strings.
  def format_earthquake_data(earthquake_res: dict) -> str: """Format an earthquake news into a readable string.""" return f""" UTC Datetime: {earthquake_res.get('utcdatetime', 'Unknown')} Local Datetime: {earthquake_res.get('localdatetime', 'Unknown')} Latitude: {earthquake_res.get('lat', 'Unknown')} Longitude: {earthquake_res.get('lon', 'Unknown')} Depth: {earthquake_res.get('depth', 'Unknown')} Location: {earthquake_res.get('location', 'Unknown')} Default Magnitude: {earthquake_res.get('magdefault', 'Unknown')} Default Magnitude Type: {earthquake_res.get('magtypedefault', 'Unknown')} Status: {earthquake_res.get('status', 'Unknown')} Distance/direction to a Malaysian location: {earthquake_res.get('n_distancemas', 'Unknown')} Distance/direction to a non-Malaysian location: {earthquake_res.get('n_distancerest', 'Unknown')} """
• weather.py:118-118 (registration)

  The @mcp.tool() decorator registers the get_earthquake_news function as an MCP tool, with schema inferred from the function signature and docstring.
  @mcp.tool()
• helpers.py:76-86 (helper)

  General helper function used to make HTTP requests to the government API endpoints.
  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:103-106 (helper)

  Helper function used to get the current date for defaulting date ranges.
  def current_date() -> str: """Get current date in YYYY-MM-DD format.""" return datetime.now().strftime("%Y-%m-%d")
• helpers.py:95-101 (helper)

  Helper function used to validate datetime inputs in the correct format.
  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