Skip to content

async_buoy_client

Async client for KMA Marine Meteorological Buoy API.

This module provides access to marine observation data from buoys and wave buoys, including wave height, water temperature, wind, and atmospheric data.

AsyncBuoyClient

Async client for accessing KMA Marine Meteorological Buoy data.

Provides access to marine observation data including: - Wave height, period, and direction - Water temperature - Wind direction and speed - Atmospheric pressure and humidity

Source code in python/src/kma_mcp/marine/async_buoy_client.py 
 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
class AsyncBuoyClient:
    """Async client for accessing KMA Marine Meteorological Buoy data.

    Provides access to marine observation data including:
    - Wave height, period, and direction
    - Water temperature
    - Wind direction and speed
    - Atmospheric pressure and humidity
    """

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

    def __init__(self, auth_key: str, timeout: float = 30.0):
        """Initialize the Buoy 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) -> 'AsyncBuoyClient':
        """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_buoy_data(self, tm: str | datetime, stn: int | str = 0) -> dict[str, Any]:
        """Get marine buoy observation data for a specific time.

        Args:
            tm: Observation time in 'YYYYMMDDHHmm' format or datetime object
            stn: Station number (0 for all stations, default: 0)

        Returns:
            Marine buoy observation data

        Example:
            >>> async with AsyncBuoyClient('your_api_key')
            >>> ...     data = await client.get_buoy_data('202501011200', stn=0)
        """
        if isinstance(tm, datetime):
            tm = tm.strftime('%Y%m%d%H%M')

        params = {'tm': tm, 'stn': str(stn), 'help': '0'}
        return await self._make_request('kma_buoy.php', params)

    async def get_buoy_period(
        self, tm1: str | datetime, tm2: str | datetime, stn: int | str = 0
    ) -> dict[str, Any]:
        """Get marine buoy observation data for a time period.

        Args:
            tm1: Start time in 'YYYYMMDDHHmm' format or datetime object
            tm2: End time in 'YYYYMMDDHHmm' format or datetime object
            stn: Station number (0 for all stations, default: 0)

        Returns:
            Marine buoy observation data for the period

        Example:
            >>> async with AsyncBuoyClient('your_api_key')
            >>> ...     data = await client.get_buoy_period('202501010000', '202501020000', stn=0)
        """
        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, 'stn': str(stn), 'help': '0'}
        return await self._make_request('kma_buoy2.php', params)

    async def get_comprehensive_marine_data(
        self, tm: str | datetime, stn: int | str = 0
    ) -> dict[str, Any]:
        """Get comprehensive marine observation data.

        Includes wave height, period, direction, water temperature,
        wind data from dual sensors, and atmospheric data.

        Args:
            tm: Observation time in 'YYYYMMDDHHmm' format or datetime object
            stn: Station number (0 for all stations, default: 0)

        Returns:
            Comprehensive marine observation data

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

        params = {'tm': tm, 'stn': str(stn), 'help': '0'}
        return await self._make_request('sea_obs.php', params)

close() async

Close the HTTP client.

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

get_buoy_data(tm, stn=0) async

Get marine buoy observation data for a specific time.

Parameters:

  • tm (str | datetime) –

    Observation time in 'YYYYMMDDHHmm' format or datetime object

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

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

Returns:

Example

async with AsyncBuoyClient('your_api_key') ... data = await client.get_buoy_data('202501011200', stn=0)

Source code in python/src/kma_mcp/marine/async_buoy_client.py 
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
async def get_buoy_data(self, tm: str | datetime, stn: int | str = 0) -> dict[str, Any]:
    """Get marine buoy observation data for a specific time.

    Args:
        tm: Observation time in 'YYYYMMDDHHmm' format or datetime object
        stn: Station number (0 for all stations, default: 0)

    Returns:
        Marine buoy observation data

    Example:
        >>> async with AsyncBuoyClient('your_api_key')
        >>> ...     data = await client.get_buoy_data('202501011200', stn=0)
    """
    if isinstance(tm, datetime):
        tm = tm.strftime('%Y%m%d%H%M')

    params = {'tm': tm, 'stn': str(stn), 'help': '0'}
    return await self._make_request('kma_buoy.php', params)

get_buoy_period(tm1, tm2, stn=0) async

Get marine buoy observation data 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

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

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

Returns:

  • dict[str, Any] –

    Marine buoy observation data for the period

Example

async with AsyncBuoyClient('your_api_key') ... data = await client.get_buoy_period('202501010000', '202501020000', stn=0)

Source code in python/src/kma_mcp/marine/async_buoy_client.py 
 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_buoy_period(
    self, tm1: str | datetime, tm2: str | datetime, stn: int | str = 0
) -> dict[str, Any]:
    """Get marine buoy observation data for a time period.

    Args:
        tm1: Start time in 'YYYYMMDDHHmm' format or datetime object
        tm2: End time in 'YYYYMMDDHHmm' format or datetime object
        stn: Station number (0 for all stations, default: 0)

    Returns:
        Marine buoy observation data for the period

    Example:
        >>> async with AsyncBuoyClient('your_api_key')
        >>> ...     data = await client.get_buoy_period('202501010000', '202501020000', stn=0)
    """
    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, 'stn': str(stn), 'help': '0'}
    return await self._make_request('kma_buoy2.php', params)

get_comprehensive_marine_data(tm, stn=0) async

Get comprehensive marine observation data.

Includes wave height, period, direction, water temperature, wind data from dual sensors, and atmospheric data.

Parameters:

  • tm (str | datetime) –

    Observation time in 'YYYYMMDDHHmm' format or datetime object

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

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

Returns:

  • dict[str, Any] –

    Comprehensive marine observation data

Example

async with AsyncBuoyClient('your_api_key') ... data = await client.get_comprehensive_marine_data('202501011200')

Source code in python/src/kma_mcp/marine/async_buoy_client.py 
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
async def get_comprehensive_marine_data(
    self, tm: str | datetime, stn: int | str = 0
) -> dict[str, Any]:
    """Get comprehensive marine observation data.

    Includes wave height, period, direction, water temperature,
    wind data from dual sensors, and atmospheric data.

    Args:
        tm: Observation time in 'YYYYMMDDHHmm' format or datetime object
        stn: Station number (0 for all stations, default: 0)

    Returns:
        Comprehensive marine observation data

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

    params = {'tm': tm, 'stn': str(stn), 'help': '0'}
    return await self._make_request('sea_obs.php', params)