Skip to content

async_amos_client

Async client for KMA Aviation Meteorology AMOS API.

AMOS (Aerodrome Meteorological Observation Station) provides weather observations from airports and aerodromes for aviation safety.

AsyncAMOSClient

Async client for accessing KMA Aviation Meteorology AMOS data.

Provides access to: - Airport weather observations - Aerodrome meteorological data - Aviation-specific weather parameters

Source code in python/src/kma_mcp/aviation/async_amos_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 AsyncAMOSClient:
    """Async client for accessing KMA Aviation Meteorology AMOS data.

    Provides access to:
    - Airport weather observations
    - Aerodrome meteorological data
    - Aviation-specific weather parameters
    """

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

    def __init__(self, auth_key: str, timeout: float = 30.0):
        """Initialize AMOS 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) -> 'AsyncAMOSClient':
        """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_airport_observations(
        self,
        tm: str,
        dtm: int = 60,
    ) -> dict[str, Any]:
        """Get aerodrome meteorological observations.

        AMOS provides weather observations from airports and aerodromes
        for aviation operations and safety.

        Args:
            tm: Observation time in 'YYYYMMDDHHmm' format
            dtm: Data time range in minutes before tm (default: 60)
                 Typical values: 30, 60, 120, 180

        Returns:
            Airport weather observation data

        Example:
            >>> with AMOSClient('api_key') as client:
            >>>     data = client.get_airport_observations('202501011200', dtm=60)
        """
        params = {'tm': tm, 'dtm': dtm, 'help': '0'}
        return await self._make_request('amos.php', params)

    async def get_amdar_data(
        self,
        tm1: str,
        tm2: str,
        st: str = 'E',
    ) -> dict[str, Any]:
        """Get AMDAR aircraft meteorological data.

        AMDAR (Aircraft Meteorological Data Relay) provides in-flight
        weather observations from commercial aircraft equipped with
        meteorological sensors.

        Args:
            tm1: Start time in 'YYYYMMDDHHmm' format
            tm2: End time in 'YYYYMMDDHHmm' format
            st: Station type filter (default: 'E')
                Options: 'E' (all), specific station codes

        Returns:
            AMDAR aircraft meteorological data

        Example:
            >>> with AMOSClient('api_key') as client:
            >>>     data = client.get_amdar_data(
            >>>         '202501011200', '202501011400', st='E'
            >>>     )
        """
        params = {'tm1': tm1, 'tm2': tm2, 'st': st, 'help': '0'}
        return await self._make_request('amdar_kma.php', params)

close() async

Close the HTTP client.

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

get_airport_observations(tm, dtm=60) async

Get aerodrome meteorological observations.

AMOS provides weather observations from airports and aerodromes for aviation operations and safety.

Parameters:

  • tm (str) –

    Observation time in 'YYYYMMDDHHmm' format

  • dtm (int, default: 60 ) –

    Data time range in minutes before tm (default: 60) Typical values: 30, 60, 120, 180

Returns:

  • dict[str, Any] –

    Airport weather observation data

Example

with AMOSClient('api_key') as client: data = client.get_airport_observations('202501011200', dtm=60)

Source code in python/src/kma_mcp/aviation/async_amos_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
async def get_airport_observations(
    self,
    tm: str,
    dtm: int = 60,
) -> dict[str, Any]:
    """Get aerodrome meteorological observations.

    AMOS provides weather observations from airports and aerodromes
    for aviation operations and safety.

    Args:
        tm: Observation time in 'YYYYMMDDHHmm' format
        dtm: Data time range in minutes before tm (default: 60)
             Typical values: 30, 60, 120, 180

    Returns:
        Airport weather observation data

    Example:
        >>> with AMOSClient('api_key') as client:
        >>>     data = client.get_airport_observations('202501011200', dtm=60)
    """
    params = {'tm': tm, 'dtm': dtm, 'help': '0'}
    return await self._make_request('amos.php', params)

get_amdar_data(tm1, tm2, st='E') async

Get AMDAR aircraft meteorological data.

AMDAR (Aircraft Meteorological Data Relay) provides in-flight weather observations from commercial aircraft equipped with meteorological sensors.

Parameters:

  • tm1 (str) –

    Start time in 'YYYYMMDDHHmm' format

  • tm2 (str) –

    End time in 'YYYYMMDDHHmm' format

  • st (str, default: 'E' ) –

    Station type filter (default: 'E') Options: 'E' (all), specific station codes

Returns:

  • dict[str, Any] –

    AMDAR aircraft meteorological data

Example

with AMOSClient('api_key') as client: data = client.get_amdar_data( '202501011200', '202501011400', st='E' )

Source code in python/src/kma_mcp/aviation/async_amos_client.py 
 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
async def get_amdar_data(
    self,
    tm1: str,
    tm2: str,
    st: str = 'E',
) -> dict[str, Any]:
    """Get AMDAR aircraft meteorological data.

    AMDAR (Aircraft Meteorological Data Relay) provides in-flight
    weather observations from commercial aircraft equipped with
    meteorological sensors.

    Args:
        tm1: Start time in 'YYYYMMDDHHmm' format
        tm2: End time in 'YYYYMMDDHHmm' format
        st: Station type filter (default: 'E')
            Options: 'E' (all), specific station codes

    Returns:
        AMDAR aircraft meteorological data

    Example:
        >>> with AMOSClient('api_key') as client:
        >>>     data = client.get_amdar_data(
        >>>         '202501011200', '202501011400', st='E'
        >>>     )
    """
    params = {'tm1': tm1, 'tm2': tm2, 'st': st, 'help': '0'}
    return await self._make_request('amdar_kma.php', params)