Skip to content

async_satellite_client

Async client for KMA Satellite Data API.

This module provides access to GK2A satellite imagery and data products.

AsyncSatelliteClient

Async client for accessing KMA GK2A Satellite data.

Provides access to: - Satellite file listings - Level 1B and Level 2 products - Various satellite imagery channels

Source code in python/src/kma_mcp/satellite/async_satellite_client.py 
 11
 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
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
class AsyncSatelliteClient:
    """Async client for accessing KMA GK2A Satellite data.

    Provides access to:
    - Satellite file listings
    - Level 1B and Level 2 products
    - Various satellite imagery channels
    """

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

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

        Args:
            auth_key: KMA API authentication key
            timeout: Request timeout in seconds (default: 60.0, longer for large files)
        """
        self.auth_key = auth_key
        self._client = httpx.AsyncClient(timeout=timeout)

    async def __aenter__(self) -> 'AsyncSatelliteClient':
        """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.

        Args:
            endpoint: API endpoint name
            params: Query parameters

        Returns:
            JSON response as dictionary
        """
        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_satellite_file_list(
        self,
        sat: str = 'GK2A',
        vars: str = 'L1B',  # noqa: A002
        area: str = 'FD',
        fmt: str = 'NetCDF',
        tm: str | None = None,
    ) -> dict[str, Any]:
        """Get list of available satellite files.

        Args:
            sat: Satellite identifier (default: 'GK2A')
            vars: Variable/product type (default: 'L1B')
                  Options: L1B, L2, etc.
            area: Region code (default: 'FD' for Full Disk)
                  Options: FD, KO (Korea), EA (East Asia), etc.
            fmt: File format (default: 'NetCDF')
            tm: Time filter in 'YYYYMMDDHHmm' format (optional)

        Returns:
            List of available satellite files

        Example:
            >>> async with AsyncSatelliteClient('your_api_key')
            >>> files = client.get_satellite_file_list(area='KO')
        """
        params: dict[str, Any] = {
            'sat': sat,
            'vars': vars,
            'area': area,
            'fmt': fmt,
            'help': '0',
        }
        if tm:
            params['tm'] = tm

        return await self._make_request('sat_file_list.php', params)

    async def get_satellite_imagery(
        self,
        level: str,
        product: str,
        area: str,
        tm: str,
    ) -> dict[str, Any]:
        """Get satellite imagery data.

        Args:
            level: Data level ('l1b' or 'l2')
            product: Product type/channel
                     For L1B: NR016, SW038, etc. (16 channels)
                     For L2: CI (Cloud Imagery), SST, etc.
            area: Area code (FD, KO, EA, ELA, TP)
            tm: Time in 'YYYYMMDDHHmm' format

        Returns:
            Satellite imagery data

        Example:
            >>> async with AsyncSatelliteClient('your_api_key') as client:
            >>>     data = await client.get_satellite_imagery(
            >>>         'l1b', 'NR016', 'KO', '202501011200'
            >>>     )
        """
        params = {
            'lvl': level,
            'dat': product,
            'are': area,
            'tm': tm,
            'typ': 'img',
            'help': '0',
        }
        return await self._make_request('sat_file_down2.php', params)

close() async

Close the HTTP client.

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

get_satellite_file_list(sat='GK2A', vars='L1B', area='FD', fmt='NetCDF', tm=None) async

Get list of available satellite files.

Parameters:

  • sat (str, default: 'GK2A' ) –

    Satellite identifier (default: 'GK2A')

  • vars (str, default: 'L1B' ) –

    Variable/product type (default: 'L1B') Options: L1B, L2, etc.

  • area (str, default: 'FD' ) –

    Region code (default: 'FD' for Full Disk) Options: FD, KO (Korea), EA (East Asia), etc.

  • fmt (str, default: 'NetCDF' ) –

    File format (default: 'NetCDF')

  • tm (str | None, default: None ) –

    Time filter in 'YYYYMMDDHHmm' format (optional)

Returns:

  • dict[str, Any] –

    List of available satellite files

Example

async with AsyncSatelliteClient('your_api_key') files = client.get_satellite_file_list(area='KO')

Source code in python/src/kma_mcp/satellite/async_satellite_client.py 
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
async def get_satellite_file_list(
    self,
    sat: str = 'GK2A',
    vars: str = 'L1B',  # noqa: A002
    area: str = 'FD',
    fmt: str = 'NetCDF',
    tm: str | None = None,
) -> dict[str, Any]:
    """Get list of available satellite files.

    Args:
        sat: Satellite identifier (default: 'GK2A')
        vars: Variable/product type (default: 'L1B')
              Options: L1B, L2, etc.
        area: Region code (default: 'FD' for Full Disk)
              Options: FD, KO (Korea), EA (East Asia), etc.
        fmt: File format (default: 'NetCDF')
        tm: Time filter in 'YYYYMMDDHHmm' format (optional)

    Returns:
        List of available satellite files

    Example:
        >>> async with AsyncSatelliteClient('your_api_key')
        >>> files = client.get_satellite_file_list(area='KO')
    """
    params: dict[str, Any] = {
        'sat': sat,
        'vars': vars,
        'area': area,
        'fmt': fmt,
        'help': '0',
    }
    if tm:
        params['tm'] = tm

    return await self._make_request('sat_file_list.php', params)

get_satellite_imagery(level, product, area, tm) async

Get satellite imagery data.

Parameters:

  • level (str) –

    Data level ('l1b' or 'l2')

  • product (str) –

    Product type/channel For L1B: NR016, SW038, etc. (16 channels) For L2: CI (Cloud Imagery), SST, etc.

  • area (str) –

    Area code (FD, KO, EA, ELA, TP)

  • tm (str) –

    Time in 'YYYYMMDDHHmm' format

Returns:

Example

async with AsyncSatelliteClient('your_api_key') as client: data = await client.get_satellite_imagery( 'l1b', 'NR016', 'KO', '202501011200' )

Source code in python/src/kma_mcp/satellite/async_satellite_client.py 
 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
async def get_satellite_imagery(
    self,
    level: str,
    product: str,
    area: str,
    tm: str,
) -> dict[str, Any]:
    """Get satellite imagery data.

    Args:
        level: Data level ('l1b' or 'l2')
        product: Product type/channel
                 For L1B: NR016, SW038, etc. (16 channels)
                 For L2: CI (Cloud Imagery), SST, etc.
        area: Area code (FD, KO, EA, ELA, TP)
        tm: Time in 'YYYYMMDDHHmm' format

    Returns:
        Satellite imagery data

    Example:
        >>> async with AsyncSatelliteClient('your_api_key') as client:
        >>>     data = await client.get_satellite_imagery(
        >>>         'l1b', 'NR016', 'KO', '202501011200'
        >>>     )
    """
    params = {
        'lvl': level,
        'dat': product,
        'are': area,
        'tm': tm,
        'typ': 'img',
        'help': '0',
    }
    return await self._make_request('sat_file_down2.php', params)