Skip to content

async_nk_client

Async KMA North Korea Meteorological Observation API client.

This module provides a client for accessing the Korea Meteorological Administration's North Korea Meteorological Observation (북한기상관측) API.

North Korea observations provide meteorological data from weather stations in North Korea for regional weather analysis and monitoring.

AsyncNKClient

Async client for KMA North Korea Meteorological Observation API.

The North Korea observation system provides meteorological data from weather stations in North Korea for regional weather analysis, forecasting, and cross-border weather monitoring.

Source code in python/src/kma_mcp/surface/async_nk_client.py 
 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
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
class AsyncNKClient:
    """Async client for KMA North Korea Meteorological Observation API.

    The North Korea observation system provides meteorological data
    from weather stations in North Korea for regional weather analysis,
    forecasting, and cross-border weather monitoring.
    """

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

    def __init__(self, auth_key: str, timeout: float = 30.0) -> None:
        """Initialize North Korea Meteorological 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) -> 'AsyncNKClient':
        """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 North Korea Meteorological 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_hourly_data(
        self,
        tm: str | datetime,
        stn: int | str = 0,
    ) -> dict[str, Any]:
        """Get hourly North Korea meteorological observation data for a single time.

        Args:
            tm: Time in 'YYYYMMDDHHmm' format or datetime object
            stn: Station number (0 for all stations)

        Returns:
            Hourly North Korea meteorological observation data

        Example:
            >>> async with AsyncNKClient('your_auth_key')
            >>> ...     data = await client.get_hourly_data('202501011200')
            >>> # Or using datetime
            >>> from datetime import datetime
            >>> ...     data = await client.get_hourly_data(datetime(2025, 1, 1, 12, 0))
        """
        if isinstance(tm, datetime):
            tm = tm.strftime('%Y%m%d%H%M')

        params = {'tm': tm, 'stn': str(stn), 'help': '0'}
        return await self._make_request('kma_nkobs.php', params)

    async def get_hourly_period(
        self,
        tm1: str | datetime,
        tm2: str | datetime,
        stn: int | str = 0,
    ) -> dict[str, Any]:
        """Get hourly North Korea meteorological observation data for a time period.

        Args:
            tm1: Start time in 'YYYYMMDDHHmm' format or datetime object
            tm2: End time in 'YYYYMMDDHHmm' format or datetime object
            stn: Station number (0 for all stations)

        Returns:
            Hourly North Korea meteorological observation data for the period

        Example:
            >>> async with AsyncNKClient('your_auth_key')
            >>> ...     data = await client.get_hourly_period('202501010000', '202501020000')
        """
        if isinstance(tm1, datetime):
            tm1 = tm1.strftime('%Y%m%d%H%M')
        if isinstance(tm2, datetime):
            tm2 = tm2.strftime('%Y%m%d%H%M')

        params = {'tm1': tm1, 'tm2': tm2, 'stn': str(stn), 'help': '0'}
        return await self._make_request('kma_nkobs_2.php', params)

    async def get_daily_data(
        self,
        tm: str | datetime,
        stn: int | str = 0,
    ) -> dict[str, Any]:
        """Get daily North Korea meteorological observation data for a single day.

        Args:
            tm: Date in 'YYYYMMDD' format or datetime object
            stn: Station number (0 for all stations)

        Returns:
            Daily North Korea meteorological observation data

        Example:
            >>> async with AsyncNKClient('your_auth_key')
            >>> ...     data = await client.get_daily_data('20250101')
        """
        if isinstance(tm, datetime):
            tm = tm.strftime('%Y%m%d')

        params = {'tm': tm, 'stn': str(stn), 'help': '0'}
        return await self._make_request('kma_nkobs_day.php', params)

    async def get_daily_period(
        self,
        tm1: str | datetime,
        tm2: str | datetime,
        stn: int | str = 0,
    ) -> dict[str, Any]:
        """Get daily North Korea meteorological observation data for a time period.

        Args:
            tm1: Start date in 'YYYYMMDD' format or datetime object
            tm2: End date in 'YYYYMMDD' format or datetime object
            stn: Station number (0 for all stations)

        Returns:
            Daily North Korea meteorological observation data for the period

        Example:
            >>> async with AsyncNKClient('your_auth_key')
            >>> ...     data = await client.get_daily_period('20250101', '20250131')
        """
        if isinstance(tm1, datetime):
            tm1 = tm1.strftime('%Y%m%d')
        if isinstance(tm2, datetime):
            tm2 = tm2.strftime('%Y%m%d')

        params = {'tm1': tm1, 'tm2': tm2, 'stn': str(stn), 'help': '0'}
        return await self._make_request('kma_nkobs_day2.php', params)

close() async

Close the HTTP client.

Source code in python/src/kma_mcp/surface/async_nk_client.py 
45
46
47
async def close(self) -> None:
    """Close the HTTP client."""
    await self._client.aclose()

get_daily_data(tm, stn=0) async

Get daily North Korea meteorological observation data for a single day.

Parameters:

  • tm (str | datetime) –

    Date in 'YYYYMMDD' format or datetime object

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

    Station number (0 for all stations)

Returns:

  • dict[str, Any] –

    Daily North Korea meteorological observation data

Example

async with AsyncNKClient('your_auth_key') ... data = await client.get_daily_data('20250101')

Source code in python/src/kma_mcp/surface/async_nk_client.py 
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
async def get_daily_data(
    self,
    tm: str | datetime,
    stn: int | str = 0,
) -> dict[str, Any]:
    """Get daily North Korea meteorological observation data for a single day.

    Args:
        tm: Date in 'YYYYMMDD' format or datetime object
        stn: Station number (0 for all stations)

    Returns:
        Daily North Korea meteorological observation data

    Example:
        >>> async with AsyncNKClient('your_auth_key')
        >>> ...     data = await client.get_daily_data('20250101')
    """
    if isinstance(tm, datetime):
        tm = tm.strftime('%Y%m%d')

    params = {'tm': tm, 'stn': str(stn), 'help': '0'}
    return await self._make_request('kma_nkobs_day.php', params)

get_daily_period(tm1, tm2, stn=0) async

Get daily North Korea meteorological observation data for a time period.

Parameters:

  • tm1 (str | datetime) –

    Start date in 'YYYYMMDD' format or datetime object

  • tm2 (str | datetime) –

    End date in 'YYYYMMDD' format or datetime object

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

    Station number (0 for all stations)

Returns:

  • dict[str, Any] –

    Daily North Korea meteorological observation data for the period

Example

async with AsyncNKClient('your_auth_key') ... data = await client.get_daily_period('20250101', '20250131')

Source code in python/src/kma_mcp/surface/async_nk_client.py 
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
async def get_daily_period(
    self,
    tm1: str | datetime,
    tm2: str | datetime,
    stn: int | str = 0,
) -> dict[str, Any]:
    """Get daily North Korea meteorological observation data for a time period.

    Args:
        tm1: Start date in 'YYYYMMDD' format or datetime object
        tm2: End date in 'YYYYMMDD' format or datetime object
        stn: Station number (0 for all stations)

    Returns:
        Daily North Korea meteorological observation data for the period

    Example:
        >>> async with AsyncNKClient('your_auth_key')
        >>> ...     data = await client.get_daily_period('20250101', '20250131')
    """
    if isinstance(tm1, datetime):
        tm1 = tm1.strftime('%Y%m%d')
    if isinstance(tm2, datetime):
        tm2 = tm2.strftime('%Y%m%d')

    params = {'tm1': tm1, 'tm2': tm2, 'stn': str(stn), 'help': '0'}
    return await self._make_request('kma_nkobs_day2.php', params)

get_hourly_data(tm, stn=0) async

Get hourly North Korea meteorological observation data for a single time.

Parameters:

  • tm (str | datetime) –

    Time in 'YYYYMMDDHHmm' format or datetime object

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

    Station number (0 for all stations)

Returns:

  • dict[str, Any] –

    Hourly North Korea meteorological observation data

Example

async with AsyncNKClient('your_auth_key') ... data = await client.get_hourly_data('202501011200')

Or using datetime

from datetime import datetime ... data = await client.get_hourly_data(datetime(2025, 1, 1, 12, 0))

Source code in python/src/kma_mcp/surface/async_nk_client.py 
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
async def get_hourly_data(
    self,
    tm: str | datetime,
    stn: int | str = 0,
) -> dict[str, Any]:
    """Get hourly North Korea meteorological observation data for a single time.

    Args:
        tm: Time in 'YYYYMMDDHHmm' format or datetime object
        stn: Station number (0 for all stations)

    Returns:
        Hourly North Korea meteorological observation data

    Example:
        >>> async with AsyncNKClient('your_auth_key')
        >>> ...     data = await client.get_hourly_data('202501011200')
        >>> # Or using datetime
        >>> from datetime import datetime
        >>> ...     data = await client.get_hourly_data(datetime(2025, 1, 1, 12, 0))
    """
    if isinstance(tm, datetime):
        tm = tm.strftime('%Y%m%d%H%M')

    params = {'tm': tm, 'stn': str(stn), 'help': '0'}
    return await self._make_request('kma_nkobs.php', params)

get_hourly_period(tm1, tm2, stn=0) async

Get hourly North Korea meteorological observation data for a time period.

Parameters:

  • tm1 (str | datetime) –

    Start time in 'YYYYMMDDHHmm' format or datetime object

  • tm2 (str | datetime) –

    End time in 'YYYYMMDDHHmm' format or datetime object

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

    Station number (0 for all stations)

Returns:

  • dict[str, Any] –

    Hourly North Korea meteorological observation data for the period

Example

async with AsyncNKClient('your_auth_key') ... data = await client.get_hourly_period('202501010000', '202501020000')

Source code in python/src/kma_mcp/surface/async_nk_client.py 
 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
async def get_hourly_period(
    self,
    tm1: str | datetime,
    tm2: str | datetime,
    stn: int | str = 0,
) -> dict[str, Any]:
    """Get hourly North Korea meteorological observation data for a time period.

    Args:
        tm1: Start time in 'YYYYMMDDHHmm' format or datetime object
        tm2: End time in 'YYYYMMDDHHmm' format or datetime object
        stn: Station number (0 for all stations)

    Returns:
        Hourly North Korea meteorological observation data for the period

    Example:
        >>> async with AsyncNKClient('your_auth_key')
        >>> ...     data = await client.get_hourly_period('202501010000', '202501020000')
    """
    if isinstance(tm1, datetime):
        tm1 = tm1.strftime('%Y%m%d%H%M')
    if isinstance(tm2, datetime):
        tm2 = tm2.strftime('%Y%m%d%H%M')

    params = {'tm1': tm1, 'tm2': tm2, 'stn': str(stn), 'help': '0'}
    return await self._make_request('kma_nkobs_2.php', params)