first commit

.gitignore

@@ -0,0 +1 @@
+__pycache__

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": [
+                "<directory of the project>"
+            ]
+        }    
+    }
+}
+```
+
+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
+

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

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