Oceanographic Seawater Properties (TEOS-10)

Usage

1. MCP Server Definition

import asyncio
import json
from contextlib import AsyncExitStack
from mcp.client.streamable_http import streamablehttp_client
from mcp import ClientSession

class OceanGSWClient:
    """Ocean GSW (Gibbs SeaWater) MCP Client"""

    def __init__(self, server_url: str, api_key: str):
        self.server_url = server_url
        self.api_key = api_key
        self.session = None

    async def connect(self):
        print(f"Connecting to: {self.server_url}")
        try:
            self.transport = streamablehttp_client(
                url=self.server_url,
                headers={"SCP-HUB-API-KEY": self.api_key}
            )
            self._stack = AsyncExitStack()
            await self._stack.__aenter__()
            self.read, self.write, self.get_session_id = await self._stack.enter_async_context(self.transport)
            self.session_ctx = ClientSession(self.read, self.write)
            self.session = await self._stack.enter_async_context(self.session_ctx)
            await self.session.initialize()
            print("✓ connect success")
            return True
        except Exception as e:
            print(f"✗ connect failure: {e}")
            return False

    async def disconnect(self):
        """Disconnect from server"""
        try:
            if hasattr(self, '_stack'):
                await self._stack.aclose()
            print("✓ already disconnect")
        except Exception as e:
            print(f"✗ disconnect error: {e}")
    def parse_result(self, result):
        try:
            if hasattr(result, 'content') and result.content:
                content = result.content[0]
                if hasattr(content, 'text'):
                    return json.loads(content.text)
            return str(result)
        except Exception as e:
            return {"error": f"parse error: {e}", "raw": str(result)}

Usage

1. MCP Server Definition

import asyncio import json from contextlib import AsyncExitStack from mcp.client.streamable_http import streamablehttp_client from mcp import ClientSession class OceanGSWClient: """Ocean GSW (Gibbs SeaWater) MCP Client""" def __init__(self, server_url: str, api_key: str): self.server_url = server_url self.api_key = api_key self.session = None async def connect(self): print(f"Connecting to: {self.server_url}") try: self.transport = streamablehttp_client( url=self.server_url, headers={"SCP-HUB-API-KEY": self.api_key} ) self._stack = AsyncExitStack() await self._stack.__aenter__() self.read, self.write, self.get_session_id = await self._stack.enter_async_context(self.transport) self.session_ctx = ClientSession(self.read, self.write) self.session = await self._stack.enter_async_context(self.session_ctx) await self.session.initialize() print("✓ connect success") return True except Exception as e: print(f"✗ connect failure: {e}") return False async def disconnect(self): """Disconnect from server""" try: if hasattr(self, '_stack'): await self._stack.aclose() print("✓ already disconnect") except Exception as e: print(f"✗ disconnect error: {e}") def parse_result(self, result): try: if hasattr(result, 'content') and result.content: content = result.content[0] if hasattr(content, 'text'): return json.loads(content.text) return str(result) except Exception as e: return {"error": f"parse error: {e}", "raw": str(result)}

## Initialize client client = OceanGSWClient( "https://scp.intern-ai.org.cn/api/v1/mcp/34/OceanGSW-Tool", "<your-api-key>" ) if not await client.connect(): print("connection failed") exit() print("=== Oceanographic Seawater Properties (TEOS-10) ===\n") ## Input parameters for typical ocean conditions practical_salinity = 35.0 # PSU (practical salinity units) temperature = 10.0 # °C (in-situ temperature) pressure = 1000.0 # dbar (approximately 1000m depth) longitude = -30.0 # degrees E latitude = 20.0 # degrees N ## Step 1: Convert practical salinity to absolute salinity print("Step 1: Convert to Absolute Salinity") result = await client.session.call_tool( "SA_from_SP", arguments={ "SP": practical_salinity, "p": pressure, "lon": longitude, "lat": latitude } ) absolute_salinity = client.parse_result(result) print(f"Practical Salinity: {practical_salinity} PSU") print(f"Absolute Salinity: {absolute_salinity} g/kg\n") ## Step 2: Calculate seawater density print("Step 2: Calculate Seawater Density") result = await client.session.call_tool( "rho", arguments={ "SA": absolute_salinity, "CT": temperature, "p": pressure } ) density = client.parse_result(result) print(f"Density: {density} kg/m³\n") ## Step 3: Calculate speed of sound print("Step 3: Calculate Sound Speed") result = await client.session.call_tool( "sound_speed", arguments={ "SA": absolute_salinity, "CT": temperature, "p": pressure } ) sound_speed = client.parse_result(result) print(f"Sound Speed: {sound_speed} m/s\n") ## Step 4: Calculate freezing temperature print("Step 4: Calculate Freezing Temperature") result = await client.session.call_tool( "t_freezing", arguments={ "SA": absolute_salinity, "p": pressure, "saturation_fraction": 0.0 # 0 for air-free, 1 for air-saturated } ) freezing_temp = client.parse_result(result) print(f"Freezing Temperature: {freezing_temp}°C\n") ## Step 5: Calculate potential temperature print("Step 5: Calculate Potential Temperature") result = await client.session.call_tool( "pt_from_t", arguments={ "SA": absolute_salinity, "t": temperature, "p": pressure, "p_ref": 0.0 # Reference pressure (0 for surface) } ) potential_temp = client.parse_result(result) print(f"In-situ Temperature: {temperature}°C") print(f"Potential Temperature: {potential_temp}°C\n") await client.disconnect()

Oceanographic Seawater Properties