Skip to content

async_radar_client

Async KMA Weather Radar API client.

This module provides a client for accessing the Korea Meteorological Administration's Weather Radar (레이더) API for precipitation detection and monitoring.

Weather radar provides real-time precipitation patterns, intensity, and movement for nowcasting and severe weather monitoring.

AsyncRadarClient

Async client for KMA Weather Radar API.

The Weather Radar system provides real-time precipitation detection and monitoring using radar reflectivity data for nowcasting, severe weather tracking, and precipitation analysis.

Source code in python/src/kma_mcp/radar/async_radar_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
class AsyncRadarClient:
    """Async client for KMA Weather Radar API.

    The Weather Radar system provides real-time precipitation detection
    and monitoring using radar reflectivity data for nowcasting,
    severe weather tracking, and precipitation analysis.
    """

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

    def __init__(self, auth_key: str, timeout: float = 30.0) -> None:
        """Initialize Weather Radar 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) -> 'AsyncRadarClient':
        """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 Radar 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_radar_image(
        self,
        tm: str | datetime,
        radar_id: str = 'ALL',
    ) -> dict[str, Any]:
        """Get weather radar image data.

        Args:
            tm: Time in 'YYYYMMDDHHmm' format or datetime object
            radar_id: Radar station ID (default: 'ALL' for composite)

        Returns:
            Weather radar image data

        Example:
            >>> async with AsyncRadarClient('your_auth_key')
            >>> ...     data = await client.get_radar_image('202501011200')
        """
        if isinstance(tm, datetime):
            tm = tm.strftime('%Y%m%d%H%M')

        params = {'tm': tm, 'radar': radar_id, 'help': '0'}
        return await self._make_request('kma_radar.php', params)

    async def get_radar_image_sequence(
        self,
        tm1: str | datetime,
        tm2: str | datetime,
        radar_id: str = 'ALL',
    ) -> dict[str, Any]:
        """Get weather radar image sequence for a time period.

        Args:
            tm1: Start time in 'YYYYMMDDHHmm' format or datetime object
            tm2: End time in 'YYYYMMDDHHmm' format or datetime object
            radar_id: Radar station ID (default: 'ALL' for composite)

        Returns:
            Weather radar image sequence data

        Example:
            >>> async with AsyncRadarClient('your_auth_key')
            >>> ...     data = await client.get_radar_image_sequence('202501011200', '202501011300')
        """
        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, 'radar': radar_id, 'help': '0'}
        return await self._make_request('kma_radar_2.php', params)

    async def get_radar_reflectivity(
        self,
        tm: str | datetime,
        x: float,
        y: float,
    ) -> dict[str, Any]:
        """Get radar reflectivity data for a specific location.

        Args:
            tm: Time in 'YYYYMMDDHHmm' format or datetime object
            x: Longitude coordinate
            y: Latitude coordinate

        Returns:
            Radar reflectivity data

        Example:
            >>> async with AsyncRadarClient('your_auth_key')
            >>> ...     data = await client.get_radar_reflectivity('202501011200', 127.0, 37.5)
        """
        if isinstance(tm, datetime):
            tm = tm.strftime('%Y%m%d%H%M')

        params = {'tm': tm, 'x': str(x), 'y': str(y), 'help': '0'}
        return await self._make_request('kma_radar_ref.php', params)

close() async

Close the HTTP client.

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

get_radar_image(tm, radar_id='ALL') async

Get weather radar image data.

Parameters:

  • tm (str | datetime) –

    Time in 'YYYYMMDDHHmm' format or datetime object

  • radar_id (str, default: 'ALL' ) –

    Radar station ID (default: 'ALL' for composite)

Returns:

Example

async with AsyncRadarClient('your_auth_key') ... data = await client.get_radar_image('202501011200')

Source code in python/src/kma_mcp/radar/async_radar_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
async def get_radar_image(
    self,
    tm: str | datetime,
    radar_id: str = 'ALL',
) -> dict[str, Any]:
    """Get weather radar image data.

    Args:
        tm: Time in 'YYYYMMDDHHmm' format or datetime object
        radar_id: Radar station ID (default: 'ALL' for composite)

    Returns:
        Weather radar image data

    Example:
        >>> async with AsyncRadarClient('your_auth_key')
        >>> ...     data = await client.get_radar_image('202501011200')
    """
    if isinstance(tm, datetime):
        tm = tm.strftime('%Y%m%d%H%M')

    params = {'tm': tm, 'radar': radar_id, 'help': '0'}
    return await self._make_request('kma_radar.php', params)

get_radar_image_sequence(tm1, tm2, radar_id='ALL') async

Get weather radar image sequence 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

  • radar_id (str, default: 'ALL' ) –

    Radar station ID (default: 'ALL' for composite)

Returns:

  • dict[str, Any] –

    Weather radar image sequence data

Example

async with AsyncRadarClient('your_auth_key') ... data = await client.get_radar_image_sequence('202501011200', '202501011300')

Source code in python/src/kma_mcp/radar/async_radar_client.py 
 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
async def get_radar_image_sequence(
    self,
    tm1: str | datetime,
    tm2: str | datetime,
    radar_id: str = 'ALL',
) -> dict[str, Any]:
    """Get weather radar image sequence for a time period.

    Args:
        tm1: Start time in 'YYYYMMDDHHmm' format or datetime object
        tm2: End time in 'YYYYMMDDHHmm' format or datetime object
        radar_id: Radar station ID (default: 'ALL' for composite)

    Returns:
        Weather radar image sequence data

    Example:
        >>> async with AsyncRadarClient('your_auth_key')
        >>> ...     data = await client.get_radar_image_sequence('202501011200', '202501011300')
    """
    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, 'radar': radar_id, 'help': '0'}
    return await self._make_request('kma_radar_2.php', params)

get_radar_reflectivity(tm, x, y) async

Get radar reflectivity data for a specific location.

Parameters:

  • tm (str | datetime) –

    Time in 'YYYYMMDDHHmm' format or datetime object

  • x (float) –

    Longitude coordinate

  • y (float) –

    Latitude coordinate

Returns:

Example

async with AsyncRadarClient('your_auth_key') ... data = await client.get_radar_reflectivity('202501011200', 127.0, 37.5)

Source code in python/src/kma_mcp/radar/async_radar_client.py 
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
async def get_radar_reflectivity(
    self,
    tm: str | datetime,
    x: float,
    y: float,
) -> dict[str, Any]:
    """Get radar reflectivity data for a specific location.

    Args:
        tm: Time in 'YYYYMMDDHHmm' format or datetime object
        x: Longitude coordinate
        y: Latitude coordinate

    Returns:
        Radar reflectivity data

    Example:
        >>> async with AsyncRadarClient('your_auth_key')
        >>> ...     data = await client.get_radar_reflectivity('202501011200', 127.0, 37.5)
    """
    if isinstance(tm, datetime):
        tm = tm.strftime('%Y%m%d%H%M')

    params = {'tm': tm, 'x': str(x), 'y': str(y), 'help': '0'}
    return await self._make_request('kma_radar_ref.php', params)