commit 61278b132ae5ca5a49be967ba06bcbf34476488c Author: pong Date: Sun May 25 17:15:04 2025 +0800 first commit diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..ed8ebf5 --- /dev/null +++ b/.gitignore @@ -0,0 +1 @@ +__pycache__ \ No newline at end of file diff --git a/README.md b/README.md new file mode 100644 index 0000000..0fdb36a --- /dev/null +++ b/README.md @@ -0,0 +1,70 @@ +# Hong Kong Observatory MCP Server + +A Model Context Protocol (MCP) server that provides weather information of Hong Kong Observatory. + +## Features + +- Weather information (forecasts, warnings, current conditions) +- Earthquake information +- Gregorian to Lunar calendar conversion +- Rainfall data from monitoring stations + +## Data Source +This project utilizes the official HKO Open Data API: +- Weather Information base URL: https://data.weather.gov.hk/weatherAPI/opendata/weather.php +- Earthquake Information base URL: https://data.weather.gov.hk/weatherAPI/opendata/earthquake.php +- Gregorian-Lunar Calendar Conversion base URL: https://data.weather.gov.hk/weatherAPI/opendata/lunardate.php +- Rainfall in the Past Hour from Automatic Weather Station base URL: https://data.weather.gov.hk/weatherAPI/opendata/hourlyRainfall.php +- [HKO Open Data API Documentation](https://www.hko.gov.hk/en/weatherAPI/doc/files/HKO_Open_Data_API_Documentation.pdf) + +## Prerequisites +- Python 3.10 or higher + +## Installation +1. Install Python + +2. Clone this repository: +```bash +git clone https://github.com/pongiotdevelop/hko_mcp.git +``` + +## Usage +1. Add the below to your MCP configuration: +```json +{ + "mcpServers": { + "HKO_MCP": { + "command": "python", + "args": [ + "" + ] + } + } +} +``` + +2. The server provides several tools that can be used by Language Models to query HKO information: +- `get_weather_info(dataType: str, lang: str = "en")`: Get weather information from the Hong Kong Observatory. +- `get_earthquake_info(dataType: str, lang: str = "en")`: Get earthquake information from the Hong Kong Observatory. +- `Gregorian_Lunar_Calendar_Conversion(date: str)`: Convert Gregorian date to Lunar date. +- `rainfall_past_hour_from_station(lang: str = "en")`: Get rainfall past hour from the Hong Kong Observatory. + +## Testing +Run the test.py for testing the API functions of MCP server: +```bash +python test.py +``` + +## Dependencies +- `requests`: For HTTP requests +- `fastmcp`: For MCP server implementation + +## Contributing + +Contributions are welcome! Please feel free to submit a Pull Request. + +## Acknowledgments + +- Hong Kong Observatory for providing the open data API +- The MCP protocol developers + diff --git a/hko_mcp.py b/hko_mcp.py new file mode 100644 index 0000000..46e0b43 --- /dev/null +++ b/hko_mcp.py @@ -0,0 +1,112 @@ +import requests +from mcp.server.fastmcp import FastMCP + +mcp = FastMCP("hko mcp") + +@mcp.tool() +def get_weather_info(dataType: str, lang: str = "en"): + """ + Get weather information from the Hong Kong Observatory. + + Args: + dataType: The type of weather information to get. + - flw: Local Weather Forecast + - fnd: 9-day Weather Forecast + - rhrread: Current Weather Report + - warnsum: Weather Warning Summary + - warningInfo: Weather Warning Information + - swt: Special Weather Tips + lang: The language of the weather information. + - en: English + - tc: Traditional Chinese + - sc: Simplified Chinese + Returns: + The weather information by JSON format. + """ + + # http get request + # url = f"https://www.hko.gov.hk/wxinfo/fnd/fnd_{lang}.htm" + url = f"https://data.weather.gov.hk/weatherAPI/opendata/weather.php?dataType={dataType}&lang={lang}" + response = requests.get(url) + response.raise_for_status() + html = response.text + # print(html) # for check the response only + + return response.json() + + +@mcp.tool() +def get_earthquake_info(dataType: str, lang: str = "en"): + """ + Get earthquake information from the Hong Kong Observatory. + + Args: + dataType: The type of earthquake information to get. + - qem: Quick Earthquake Messages + - feltearthquake: Locally Felt Earth Tremor Report + lang: The language of the earthquake information. + - en: English + - tc: Traditional Chinese + - sc: Simplified Chinese + Returns: + The earthquake information by JSON format. + """ + + url = f"https://data.weather.gov.hk/weatherAPI/opendata/earthquake.php?dataType={dataType}&lang={lang}" + response = requests.get(url) + response.raise_for_status() + html = response.text + # print(html) # for check the response only + + return response.json() + + +@mcp.tool() +def Gregorian_Lunar_Calendar_Coversion(date: str): + """ + Convert Gregorian date to Lunar date. + + Args: + date: The Gregorian date to convert. + - Format: YYYY-MM-DD + Returns: + The Lunar date in by JSON format. + Args: + LunarYear: The Lunar year. + LunarDate: The Lunar date. + """ + + url = f"https://data.weather.gov.hk/weatherAPI/opendata/lunardate.php??date={date}" + response = requests.get(url) + response.raise_for_status() + html = response.text + # print(html) # for check the response only + + return html + + +@mcp.tool() +def rainfall_past_hour_from_station(lang: str = "en"): + """ + Get rainfall past hour from the Hong Kong Observatory. + + Args: + lang: The language of the rainfall information. + - en: English + - tc: Traditional Chinese + - sc: Simplified Chinese + Returns: + The rainfall past hour information by JSON format. + """ + + url = f"https://data.weather.gov.hk/weatherAPI/opendata/hourlyRainfall.php?lang={lang}" + response = requests.get(url) + response.raise_for_status() + html = response.text + # print(html) # for check the response only + + return response.json() + + +if __name__ == '__main__': + mcp.run(transport='stdio') \ No newline at end of file diff --git a/test.py b/test.py new file mode 100644 index 0000000..e3e99f3 --- /dev/null +++ b/test.py @@ -0,0 +1,100 @@ +import unittest +from unittest.mock import patch, MagicMock +import json +from hko_mcp import get_weather_info, get_earthquake_info, Gregorian_Lunar_Calendar_Coversion, rainfall_past_hour_from_station + +class TestHKOMCP(unittest.TestCase): + + @patch('hko_mcp.requests.get') + def test_get_weather_info(self, mock_get): + # Setup mock response for different data types + mock_response = MagicMock() + mock_response.json.return_value = {"forecast": "Sunny"} + mock_response.text = json.dumps({"forecast": "Sunny"}) + mock_get.return_value = mock_response + + # Test with different data types + data_types = ["flw", "fnd", "rhrread", "warnsum", "warningInfo", "swt"] + languages = ["en", "tc", "sc"] + + for data_type in data_types: + for lang in languages: + result = get_weather_info(data_type, lang) + self.assertEqual(result, {"forecast": "Sunny"}) + mock_get.assert_called_with( + f"https://data.weather.gov.hk/weatherAPI/opendata/weather.php?dataType={data_type}&lang={lang}" + ) + + @patch('hko_mcp.requests.get') + def test_get_earthquake_info(self, mock_get): + # Setup mock response + mock_response = MagicMock() + mock_response.json.return_value = {"earthquake": "detected"} + mock_response.text = json.dumps({"earthquake": "detected"}) + mock_get.return_value = mock_response + + # Test with different data types + data_types = ["qem", "feltearthquake"] + languages = ["en", "tc", "sc"] + + for data_type in data_types: + for lang in languages: + result = get_earthquake_info(data_type, lang) + self.assertEqual(result, {"earthquake": "detected"}) + mock_get.assert_called_with( + f"https://data.weather.gov.hk/weatherAPI/opendata/earthquake.php?dataType={data_type}&lang={lang}" + ) + + @patch('hko_mcp.requests.get') + def test_gregorian_lunar_calendar_conversion(self, mock_get): + # Setup mock response + mock_response = MagicMock() + mock_response.text = json.dumps({"LunarYear": 2023, "LunarDate": "8-16"}) + mock_get.return_value = mock_response + + # Test with a sample date + date = "2023-10-01" + result = Gregorian_Lunar_Calendar_Coversion(date) + self.assertEqual(result, json.dumps({"LunarYear": 2023, "LunarDate": "8-16"})) + mock_get.assert_called_with( + f"https://data.weather.gov.hk/weatherAPI/opendata/lunardate.php??date={date}" + ) + + @patch('hko_mcp.requests.get') + def test_rainfall_past_hour_from_station(self, mock_get): + # Setup mock response + mock_response = MagicMock() + mock_response.json.return_value = {"rainfall": {"Central": 0.5}} + mock_response.text = json.dumps({"rainfall": {"Central": 0.5}}) + mock_get.return_value = mock_response + + # Test with different languages + languages = ["en", "tc", "sc"] + + for lang in languages: + result = rainfall_past_hour_from_station(lang) + self.assertEqual(result, {"rainfall": {"Central": 0.5}}) + mock_get.assert_called_with( + f"https://data.weather.gov.hk/weatherAPI/opendata/hourlyRainfall.php?lang={lang}" + ) + + @patch('hko_mcp.requests.get') + def test_error_handling(self, mock_get): + # Setup mock response to raise an exception + mock_get.side_effect = Exception("API Error") + + # Test error handling for each function + with self.assertRaises(Exception): + get_weather_info("flw", "en") + + with self.assertRaises(Exception): + get_earthquake_info("qem", "en") + + with self.assertRaises(Exception): + Gregorian_Lunar_Calendar_Coversion("2023-10-01") + + with self.assertRaises(Exception): + rainfall_past_hour_from_station("en") + +if __name__ == '__main__': + unittest.main() \ No newline at end of file