Skip to content

async_integrated_client

Async client for KMA Integrated Meteorology API.

Provides access to integrated meteorological data combining various observation sources and specialized data products.

AsyncIntegratedClient

Async client for accessing KMA Integrated Meteorology data.

Provides access to: - Lightning detection data - Wind profiler data - Integrated observation products

Source code in python/src/kma_mcp/integrated/async_integrated_client.py 
 12
 13
 14
 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
class AsyncIntegratedClient:
    """Async client for accessing KMA Integrated Meteorology data.

    Provides access to:
    - Lightning detection data
    - Wind profiler data
    - Integrated observation products
    """

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

    def __init__(self, auth_key: str, timeout: float = 30.0):
        """Initialize Integrated client.

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

    async def __aenter__(self) -> 'AsyncIntegratedClient':
        """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 an API request."""
        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_lightning_data(
        self,
        tm1: str,
        tm2: str,
    ) -> dict[str, Any]:
        """Get lightning detection data.

        Provides lightning strike location and intensity data from
        the KMA lightning detection network.

        Args:
            tm1: Start time in 'YYYYMMDDHHmm' format
            tm2: End time in 'YYYYMMDDHHmm' format

        Returns:
            Lightning detection data including location and intensity

        Example:
            >>> with IntegratedClient('api_key') as client:
            >>>     data = client.get_lightning_data(
            >>>         '202501011200', '202501011500'
            >>>     )
        """
        params = {'tm1': tm1, 'tm2': tm2, 'help': '0'}
        return await self._make_request('lgt_kma_np3.php', params)

    async def get_wind_profiler_data(
        self,
        tm: str,
        stn: int = 0,
        mode: str = 'L',
    ) -> dict[str, Any]:
        """Get wind profiler data.

        Wind profilers provide vertical profiles of wind speed and
        direction at various altitudes.

        Args:
            tm: Observation time in 'YYYYMMDDHHmm' format
            stn: Station ID (0 for all stations, default: 0)
            mode: Data mode (default: 'L')
                  Options: 'L' (low mode), 'H' (high mode)

        Returns:
            Wind profiler vertical profile data

        Example:
            >>> with IntegratedClient('api_key') as client:
            >>>     data = client.get_wind_profiler_data(
            >>>         '202501011200', stn=0, mode='L'
            >>>     )
        """
        params = {'tm': tm, 'stn': stn, 'mode': mode, 'help': '0'}
        return await self._make_request('kma_wpf.php', params)

close() async

Close the HTTP client.

Source code in python/src/kma_mcp/integrated/async_integrated_client.py 
41
42
43
async def close(self) -> None:
    """Close the HTTP client."""
    await self._client.aclose()

get_lightning_data(tm1, tm2) async

Get lightning detection data.

Provides lightning strike location and intensity data from the KMA lightning detection network.

Parameters:

  • tm1 (str) –

    Start time in 'YYYYMMDDHHmm' format

  • tm2 (str) –

    End time in 'YYYYMMDDHHmm' format

Returns:

  • dict[str, Any] –

    Lightning detection data including location and intensity

Example

with IntegratedClient('api_key') as client: data = client.get_lightning_data( '202501011200', '202501011500' )

Source code in python/src/kma_mcp/integrated/async_integrated_client.py 
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
async def get_lightning_data(
    self,
    tm1: str,
    tm2: str,
) -> dict[str, Any]:
    """Get lightning detection data.

    Provides lightning strike location and intensity data from
    the KMA lightning detection network.

    Args:
        tm1: Start time in 'YYYYMMDDHHmm' format
        tm2: End time in 'YYYYMMDDHHmm' format

    Returns:
        Lightning detection data including location and intensity

    Example:
        >>> with IntegratedClient('api_key') as client:
        >>>     data = client.get_lightning_data(
        >>>         '202501011200', '202501011500'
        >>>     )
    """
    params = {'tm1': tm1, 'tm2': tm2, 'help': '0'}
    return await self._make_request('lgt_kma_np3.php', params)

get_wind_profiler_data(tm, stn=0, mode='L') async

Get wind profiler data.

Wind profilers provide vertical profiles of wind speed and direction at various altitudes.

Parameters:

  • tm (str) –

    Observation time in 'YYYYMMDDHHmm' format

  • stn (int, default: 0 ) –

    Station ID (0 for all stations, default: 0)

  • mode (str, default: 'L' ) –

    Data mode (default: 'L') Options: 'L' (low mode), 'H' (high mode)

Returns:

  • dict[str, Any] –

    Wind profiler vertical profile data

Example

with IntegratedClient('api_key') as client: data = client.get_wind_profiler_data( '202501011200', stn=0, mode='L' )

Source code in python/src/kma_mcp/integrated/async_integrated_client.py 
 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
async def get_wind_profiler_data(
    self,
    tm: str,
    stn: int = 0,
    mode: str = 'L',
) -> dict[str, Any]:
    """Get wind profiler data.

    Wind profilers provide vertical profiles of wind speed and
    direction at various altitudes.

    Args:
        tm: Observation time in 'YYYYMMDDHHmm' format
        stn: Station ID (0 for all stations, default: 0)
        mode: Data mode (default: 'L')
              Options: 'L' (low mode), 'H' (high mode)

    Returns:
        Wind profiler vertical profile data

    Example:
        >>> with IntegratedClient('api_key') as client:
        >>>     data = client.get_wind_profiler_data(
        >>>         '202501011200', stn=0, mode='L'
        >>>     )
    """
    params = {'tm': tm, 'stn': stn, 'mode': mode, 'help': '0'}
    return await self._make_request('kma_wpf.php', params)