get_warning
Retrieve weather warnings for Malaysia within a specified date range to monitor alerts and plan accordingly.
Instructions
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.Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| datetime_start | No | ||
| datetime_end | No |
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