Skip to content

async_station_client

Async KMA Surface Observation Station Information API client.

This module provides a client for accessing the Korea Meteorological Administration's Surface Observation Station Information (지상관측 지점정보) API.

Station information provides metadata about weather observation stations including location, altitude, and operational status.

AsyncStationClient

Async client for KMA Surface Observation Station Information API.

The Station Information system provides metadata about weather observation stations including location coordinates, altitude, station type, and operational status.

Source code in python/src/kma_mcp/surface/async_station_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
class AsyncStationClient:
    """Async client for KMA Surface Observation Station Information API.

    The Station Information system provides metadata about weather
    observation stations including location coordinates, altitude,
    station type, and operational status.
    """

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

    def __init__(self, auth_key: str, timeout: float = 30.0) -> None:
        """Initialize Station Information 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) -> 'AsyncStationClient':
        """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 Station Information 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_asos_stations(self, stn: int | str = 0) -> dict[str, Any]:
        """Get ASOS station information.

        Args:
            stn: Station number (0 for all stations)

        Returns:
            ASOS station information

        Example:
            >>> async with AsyncStationClient('your_auth_key')
            >>> ...     data = await client.get_asos_stations()  # All stations
            >>> ...     data = await client.get_asos_stations(108)  # Specific station
        """
        params = {'stn': str(stn), 'help': '0'}
        return await self._make_request('kma_stnlist.php', params)

    async def get_aws_stations(self, stn: int | str = 0) -> dict[str, Any]:
        """Get AWS station information.

        Args:
            stn: Station number (0 for all stations)

        Returns:
            AWS station information

        Example:
            >>> async with AsyncStationClient('your_auth_key')
            >>> ...     data = await client.get_aws_stations()  # All stations
        """
        params = {'stn': str(stn), 'help': '0'}
        return await self._make_request('kma_aws_stnlist.php', params)

close() async

Close the HTTP client.

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

get_asos_stations(stn=0) async

Get ASOS station information.

Parameters:

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

    Station number (0 for all stations)

Returns:

Example

async with AsyncStationClient('your_auth_key') ... data = await client.get_asos_stations() # All stations ... data = await client.get_asos_stations(108) # Specific station

Source code in python/src/kma_mcp/surface/async_station_client.py 
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
async def get_asos_stations(self, stn: int | str = 0) -> dict[str, Any]:
    """Get ASOS station information.

    Args:
        stn: Station number (0 for all stations)

    Returns:
        ASOS station information

    Example:
        >>> async with AsyncStationClient('your_auth_key')
        >>> ...     data = await client.get_asos_stations()  # All stations
        >>> ...     data = await client.get_asos_stations(108)  # Specific station
    """
    params = {'stn': str(stn), 'help': '0'}
    return await self._make_request('kma_stnlist.php', params)

get_aws_stations(stn=0) async

Get AWS station information.

Parameters:

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

    Station number (0 for all stations)

Returns:

Example

async with AsyncStationClient('your_auth_key') ... data = await client.get_aws_stations() # All stations

Source code in python/src/kma_mcp/surface/async_station_client.py 
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
async def get_aws_stations(self, stn: int | str = 0) -> dict[str, Any]:
    """Get AWS station information.

    Args:
        stn: Station number (0 for all stations)

    Returns:
        AWS station information

    Example:
        >>> async with AsyncStationClient('your_auth_key')
        >>> ...     data = await client.get_aws_stations()  # All stations
    """
    params = {'stn': str(stn), 'help': '0'}
    return await self._make_request('kma_aws_stnlist.php', params)