Skip to content

async_warning_client

Async KMA Weather Warning API client.

This module provides a client for accessing the Korea Meteorological Administration's Weather Warning (특보) API for weather alerts and warnings.

Weather warnings provide critical alerts for severe weather conditions including heavy rain, strong winds, heavy snow, and other hazards.

AsyncWarningClient

Async client for KMA Weather Warning API.

The Weather Warning system provides alerts for severe weather conditions to protect life and property from meteorological hazards.

Source code in python/src/kma_mcp/forecast/async_warning_client.py 
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
class AsyncWarningClient:
    """Async client for KMA Weather Warning API.

    The Weather Warning system provides alerts for severe weather
    conditions to protect life and property from meteorological hazards.
    """

    BASE_URL = 'https://apihub.kma.go.kr/api/typ01/url'

    def __init__(self, auth_key: str, timeout: float = 30.0) -> None:
        """Initialize Weather Warning client.

        Args:
            auth_key: KMA API authentication key
            timeout: Request timeout in seconds (default: 30.0)
        """
        self.auth_key = auth_key
        self.timeout = timeout
        self._client = httpx.AsyncClient(timeout=timeout)

    async def __aenter__(self) -> 'AsyncWarningClient':
        """Async context manager entry."""
        return self

    async def __aexit__(self, *args: object) -> None:
        """Async context manager exit."""
        await self.close()

    async def close(self) -> None:
        """Close the HTTP client."""
        await self._client.aclose()

    async def _make_request(self, endpoint: str, params: dict[str, Any]) -> dict[str, Any]:
        """Make HTTP request to Weather Warning API.

        Args:
            endpoint: API endpoint path
            params: Query parameters

        Returns:
            API response as dictionary

        Raises:
            httpx.HTTPError: If request fails
        """
        params['authKey'] = self.auth_key
        url = f'{self.BASE_URL}/{endpoint}'
        response = await self._client.get(url, params=params)
        response.raise_for_status()
        return response.json()

    async def get_current_warnings(
        self,
        stn: int | str = 0,
    ) -> dict[str, Any]:
        """Get current active weather warnings.

        Args:
            stn: Station/region code (0 for all regions)

        Returns:
            Current active weather warnings

        Example:
            >>> async with AsyncWarningClient('your_auth_key')
            >>> ...     data = await client.get_current_warnings()
        """
        params = {'stn': str(stn), 'help': '0'}
        return await self._make_request('kma_wn.php', params)

    async def get_warning_history(
        self,
        start_date: str,
        end_date: str,
        stn: int | str = 0,
    ) -> dict[str, Any]:
        """Get weather warning history for a date range.

        Args:
            start_date: Start date in 'YYYYMMDD' format
            end_date: End date in 'YYYYMMDD' format
            stn: Station/region code (0 for all regions)

        Returns:
            Weather warning history

        Example:
            >>> async with AsyncWarningClient('your_auth_key')
            >>> ...     data = await client.get_warning_history('20250101', '20250131')
        """
        params = {'tm1': start_date, 'tm2': end_date, 'stn': str(stn), 'help': '0'}
        return await self._make_request('kma_wn_2.php', params)

    async def get_special_weather_report(
        self,
        tm: str,
        stn: int | str = 0,
    ) -> dict[str, Any]:
        """Get special weather report.

        Args:
            tm: Time in 'YYYYMMDDHHmm' format
            stn: Station/region code (0 for all regions)

        Returns:
            Special weather report data

        Example:
            >>> async with AsyncWarningClient('your_auth_key')
            >>> ...     data = await client.get_special_weather_report('202501011200')
        """
        params = {'tm': tm, 'stn': str(stn), 'help': '0'}
        return await self._make_request('kma_swr.php', params)

close() async

Close the HTTP client.

Source code in python/src/kma_mcp/forecast/async_warning_client.py 
43
44
45
async def close(self) -> None:
    """Close the HTTP client."""
    await self._client.aclose()

get_current_warnings(stn=0) async

Get current active weather warnings.

Parameters:

  • stn (int | str, default: 0 ) –

    Station/region code (0 for all regions)

Returns:

  • dict[str, Any] –

    Current active weather warnings

Example

async with AsyncWarningClient('your_auth_key') ... data = await client.get_current_warnings()

Source code in python/src/kma_mcp/forecast/async_warning_client.py 
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
async def get_current_warnings(
    self,
    stn: int | str = 0,
) -> dict[str, Any]:
    """Get current active weather warnings.

    Args:
        stn: Station/region code (0 for all regions)

    Returns:
        Current active weather warnings

    Example:
        >>> async with AsyncWarningClient('your_auth_key')
        >>> ...     data = await client.get_current_warnings()
    """
    params = {'stn': str(stn), 'help': '0'}
    return await self._make_request('kma_wn.php', params)

get_special_weather_report(tm, stn=0) async

Get special weather report.

Parameters:

  • tm (str) –

    Time in 'YYYYMMDDHHmm' format

  • stn (int | str, default: 0 ) –

    Station/region code (0 for all regions)

Returns:

Example

async with AsyncWarningClient('your_auth_key') ... data = await client.get_special_weather_report('202501011200')

Source code in python/src/kma_mcp/forecast/async_warning_client.py 
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
async def get_special_weather_report(
    self,
    tm: str,
    stn: int | str = 0,
) -> dict[str, Any]:
    """Get special weather report.

    Args:
        tm: Time in 'YYYYMMDDHHmm' format
        stn: Station/region code (0 for all regions)

    Returns:
        Special weather report data

    Example:
        >>> async with AsyncWarningClient('your_auth_key')
        >>> ...     data = await client.get_special_weather_report('202501011200')
    """
    params = {'tm': tm, 'stn': str(stn), 'help': '0'}
    return await self._make_request('kma_swr.php', params)

get_warning_history(start_date, end_date, stn=0) async

Get weather warning history for a date range.

Parameters:

  • start_date (str) –

    Start date in 'YYYYMMDD' format

  • end_date (str) –

    End date in 'YYYYMMDD' format

  • stn (int | str, default: 0 ) –

    Station/region code (0 for all regions)

Returns:

Example

async with AsyncWarningClient('your_auth_key') ... data = await client.get_warning_history('20250101', '20250131')

Source code in python/src/kma_mcp/forecast/async_warning_client.py 
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
async def get_warning_history(
    self,
    start_date: str,
    end_date: str,
    stn: int | str = 0,
) -> dict[str, Any]:
    """Get weather warning history for a date range.

    Args:
        start_date: Start date in 'YYYYMMDD' format
        end_date: End date in 'YYYYMMDD' format
        stn: Station/region code (0 for all regions)

    Returns:
        Weather warning history

    Example:
        >>> async with AsyncWarningClient('your_auth_key')
        >>> ...     data = await client.get_warning_history('20250101', '20250131')
    """
    params = {'tm1': start_date, 'tm2': end_date, 'stn': str(stn), 'help': '0'}
    return await self._make_request('kma_wn_2.php', params)