MCP/mcnoaa-tides
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Add tool annotations, structured logging, sampling summaries, and client diagnostics

Browse Source 
- ToolAnnotations (readOnlyHint, openWorldHint) on all 14 tools
- ctx.info/warning/debug logging for data quality, fetch status, assessments
- ctx.sample() summaries in conditions snapshot, deployment briefing, anomaly detection
- New test_client_capabilities diagnostic tool for MCP feature discovery
This commit is contained in:
Ryan Malloy 2026-02-23 15:01:07 -07:00
parent fb574d26b8
commit 2443d0687a
12 changed files with 285 additions and 24 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

11
src/mcnoaa_tides/server.py
View File

@ -8,7 +8,15 @@ from fastmcp import FastMCP
from mcnoaa_tides import __version__, prompts, resources
from mcnoaa_tides.client import NOAAClient
from mcnoaa_tides.tools import charts, conditions, meteorological, smartpot, stations, tides
from mcnoaa_tides.tools import (
charts,
conditions,
diagnostics,
meteorological,
smartpot,
stations,
tides,
)
@asynccontextmanager
@ -52,6 +60,7 @@ meteorological.register(mcp)
conditions.register(mcp)
smartpot.register(mcp)
charts.register(mcp)
diagnostics.register(mcp)
# Register resources and prompts
resources.register(mcp)

8
src/mcnoaa_tides/tools/charts.py
View File

@ -7,13 +7,16 @@ from typing import Literal
from fastmcp import Context, FastMCP
from fastmcp.utilities.types import Image
from mcp.types import ToolAnnotations
from mcnoaa_tides.charts import check_deps
from mcnoaa_tides.client import NOAAClient
_ANNOTATIONS = ToolAnnotations(readOnlyHint=True, openWorldHint=True)
def register(mcp: FastMCP) -> None:
@mcp.tool(tags={"visualization"})
@mcp.tool(tags={"visualization"}, annotations=_ANNOTATIONS)
async def visualize_tides(
ctx: Context,
station_id: str,
@ -70,6 +73,7 @@ def register(mcp: FastMCP) -> None:
)
observed = obs_raw.get("data", [])
except Exception:
await ctx.debug("Observed data unavailable — rendering predictions only")
pass # observed overlay is optional — skip on failure
# Look up station name
@ -96,7 +100,7 @@ def register(mcp: FastMCP) -> None:
path = _save_html(html, station_id, "tides")
return f"Interactive tide chart saved to: {path}"
@mcp.tool(tags={"visualization"})
@mcp.tool(tags={"visualization"}, annotations=_ANNOTATIONS)
async def visualize_conditions(
ctx: Context,
station_id: str,

47
src/mcnoaa_tides/tools/conditions.py
View File

@ -1,15 +1,19 @@
"""Marine conditions snapshot — parallel multi-product fetch."""
import asyncio
import json
from datetime import datetime, timezone
from fastmcp import Context, FastMCP
from mcp.types import ToolAnnotations
from mcnoaa_tides.client import NOAAClient
_ANNOTATIONS = ToolAnnotations(readOnlyHint=True, openWorldHint=True)
def register(mcp: FastMCP) -> None:
@mcp.tool(tags={"planning"})
@mcp.tool(tags={"planning"}, annotations=_ANNOTATIONS)
async def marine_conditions_snapshot(
ctx: Context,
station_id: str,
@ -57,6 +61,7 @@ def register(mcp: FastMCP) -> None:
result = name, data
except Exception as exc:
msg = str(exc) or type(exc).__name__
await ctx.warning(f"Failed to fetch {name}: {type(exc).__name__}: {msg}")
result = name, f"{type(exc).__name__}: {msg}"
completed += 1
await ctx.report_progress(completed, total, f"Fetched {name}")
@ -80,5 +85,45 @@ def register(mcp: FastMCP) -> None:
if unavailable:
snapshot["unavailable"] = unavailable
await ctx.warning(
f"Station {station_id}: {len(unavailable)} product(s) unavailable: "
+ ", ".join(unavailable.keys())
)
# Attempt to generate a natural-language summary via sampling
summary_input: dict = {"station_id": station_id}
if "predictions" in snapshot:
summary_input["tide_events"] = snapshot["predictions"].get("predictions", [])[:6]
if "wind" in snapshot:
wind_readings = snapshot["wind"].get("data", [])
if wind_readings:
summary_input["latest_wind"] = wind_readings[-1]
if "air_temperature" in snapshot:
air_readings = snapshot["air_temperature"].get("data", [])
if air_readings:
summary_input["latest_air_temp"] = air_readings[-1]
if "water_temperature" in snapshot:
water_readings = snapshot["water_temperature"].get("data", [])
if water_readings:
summary_input["latest_water_temp"] = water_readings[-1]
if "air_pressure" in snapshot:
pressure_readings = snapshot["air_pressure"].get("data", [])
if pressure_readings:
summary_input["latest_pressure"] = pressure_readings[-1]
try:
sampling_result = await ctx.sample(
f"Summarize these marine conditions in 2-3 concise sentences "
f"for a boat captain:\n{json.dumps(summary_input)}",
system_prompt=(
"You are a marine weather briefer. Be concise, factual, and actionable. "
"Mention wind, tide timing, temperature, and any concerns."
),
max_tokens=256,
)
if sampling_result.text:
snapshot["summary"] = sampling_result.text
except Exception:
pass # Client doesn't support sampling — that's fine
return snapshot

76
src/mcnoaa_tides/tools/diagnostics.py Normal file
View File

@ -0,0 +1,76 @@
"""MCP client capability diagnostics."""
from fastmcp import Context, FastMCP
from mcp.types import ToolAnnotations
_ANNOTATIONS = ToolAnnotations(readOnlyHint=True, openWorldHint=False)
def register(mcp: FastMCP) -> None:
@mcp.tool(tags={"diagnostics"}, annotations=_ANNOTATIONS)
async def test_client_capabilities(ctx: Context) -> dict:
"""Report which MCP features the connected client supports.
Returns the client implementation info, protocol version,
and capability flags for sampling, elicitation, roots, and tasks.
Useful for understanding what features are available before
calling tools that depend on optional client capabilities.
"""
result: dict = {
"session_id": ctx.session_id,
}
client_params = ctx.session.client_params
if client_params is None:
result["error"] = "No client params available"
return result
# Client implementation info
result["client"] = {
"name": client_params.clientInfo.name,
"version": client_params.clientInfo.version,
}
result["protocol_version"] = client_params.protocolVersion
# Capability flags
caps = client_params.capabilities
result["capabilities"] = {
"sampling": caps.sampling is not None,
"elicitation": caps.elicitation is not None,
"roots": caps.roots is not None,
"tasks": caps.tasks is not None,
}
# Sampling detail
if caps.sampling:
result["sampling_detail"] = {
"context": caps.sampling.context is not None,
"tools": caps.sampling.tools is not None,
}
# Elicitation detail
if caps.elicitation:
result["elicitation_detail"] = {
"form": caps.elicitation.form is not None,
"url": caps.elicitation.url is not None,
}
# Roots detail
if caps.roots:
result["roots_detail"] = {
"list_changed": bool(caps.roots.listChanged),
}
# Tasks detail
if caps.tasks:
result["tasks_detail"] = {
"list": caps.tasks.list is not None,
"cancel": caps.tasks.cancel is not None,
"requests": caps.tasks.requests is not None,
}
# Experimental / extensions
if caps.experimental:
result["experimental"] = list(caps.experimental.keys())
return result

6
src/mcnoaa_tides/tools/meteorological.py
View File

@ -3,9 +3,12 @@
from typing import Literal
from fastmcp import Context, FastMCP
from mcp.types import ToolAnnotations
from mcnoaa_tides.client import NOAAClient
_ANNOTATIONS = ToolAnnotations(readOnlyHint=True, openWorldHint=True)
MetProduct = Literal[
"air_temperature",
"water_temperature",
@ -19,7 +22,7 @@ MetProduct = Literal[
def register(mcp: FastMCP) -> None:
@mcp.tool(tags={"weather"})
@mcp.tool(tags={"weather"}, annotations=_ANNOTATIONS)
async def get_meteorological_data(
ctx: Context,
station_id: str,
@ -46,6 +49,7 @@ def register(mcp: FastMCP) -> None:
Date format: yyyyMMdd or "yyyyMMdd HH:mm".
"""
noaa: NOAAClient = ctx.lifespan_context["noaa_client"]
await ctx.info(f"Fetching {product} for station {station_id}")
return await noaa.get_data(
station_id,
product=product,

89
src/mcnoaa_tides/tools/smartpot.py
View File

@ -6,9 +6,11 @@ anomaly detection.
"""
import asyncio
import json
from datetime import datetime, timedelta, timezone
from fastmcp import Context, FastMCP
from mcp.types import ToolAnnotations
from mcnoaa_tides.client import NOAAClient
from mcnoaa_tides.tidal import (
@ -17,6 +19,8 @@ from mcnoaa_tides.tidal import (
parse_hilo_predictions,
)
_ANNOTATIONS = ToolAnnotations(readOnlyHint=True, openWorldHint=True)
async def _resolve_station(
noaa: NOAAClient,
@ -55,7 +59,7 @@ async def _resolve_station(
def register(mcp: FastMCP) -> None:
@mcp.tool(tags={"smartpot"})
@mcp.tool(tags={"smartpot"}, annotations=_ANNOTATIONS)
async def tidal_phase(
ctx: Context,
station_id: str = "",
@ -124,9 +128,10 @@ def register(mcp: FastMCP) -> None:
}
if distance_nm is not None:
result["station_distance_nm"] = distance_nm
await ctx.debug(f"Tidal phase at {station['name']}: {phase_info.get('phase', 'unknown')}")
return result
@mcp.tool(tags={"smartpot"})
@mcp.tool(tags={"smartpot"}, annotations=_ANNOTATIONS)
async def deployment_briefing(
ctx: Context,
latitude: float,
@ -248,7 +253,21 @@ def register(mcp: FastMCP) -> None:
f"Pressure dropping {drop:.1f} mb — possible approaching weather system"
)
return {
# Log assessment level
if assessment == "NO-GO":
await ctx.warning(
f"Deployment NO-GO at {station['name']}: "
+ "; ".join(advisories)
)
elif assessment == "CAUTION":
await ctx.warning(
f"Deployment CAUTION at {station['name']}: "
+ "; ".join(advisories)
)
else:
await ctx.info(f"Deployment GO at {station['name']}")
result = {
"station": station,
"station_distance_nm": distance_nm,
"timestamp_utc": now_utc.isoformat(),
@ -264,7 +283,34 @@ def register(mcp: FastMCP) -> None:
"advisories": advisories,
}
@mcp.tool(tags={"smartpot"})
# Attempt to generate a natural-language briefing via sampling
try:
briefing_input = {
"assessment": assessment,
"advisories": advisories,
"station": station["name"],
"soak_hours": soak_hours,
"tidal_cycles": tidal_cycles,
"conditions": conditions,
}
sampling_result = await ctx.sample(
f"Write a concise deployment briefing paragraph for a crab pot "
f"crew based on this data:\n{json.dumps(briefing_input)}",
system_prompt=(
"You are a crab pot deployment advisor. Be concise, factual, "
"and actionable. State the GO/CAUTION/NO-GO assessment, key "
"conditions, tide count, and any advisories in 2-3 sentences."
),
max_tokens=256,
)
if sampling_result.text:
result["summary"] = sampling_result.text
except Exception:
pass # Client doesn't support sampling
return result
@mcp.tool(tags={"smartpot"}, annotations=_ANNOTATIONS)
async def catch_tidal_context(
ctx: Context,
events: list[dict],
@ -376,7 +422,7 @@ def register(mcp: FastMCP) -> None:
return enriched
@mcp.tool(tags={"smartpot"})
@mcp.tool(tags={"smartpot"}, annotations=_ANNOTATIONS)
async def water_level_anomaly(
ctx: Context,
station_id: str,
@ -488,7 +534,13 @@ def register(mcp: FastMCP) -> None:
"conditions as expected"
)
return {
if risk in ("high", "elevated"):
await ctx.warning(
f"Station {station_id}: {risk} anomaly — "
f"max {max_dev:.2f} ft {direction} predictions"
)
result = {
"station_id": station_id,
"window_hours": window_hours,
"threshold_ft": threshold_ft,
@ -499,3 +551,28 @@ def register(mcp: FastMCP) -> None:
"direction": direction,
"sample_count": len(deviations),
}
# Attempt to generate a concise risk summary via sampling
try:
anomaly_input = {
"station_id": station_id,
"risk_level": risk,
"max_deviation_ft": round(max_dev, 3),
"direction": direction,
"window_hours": window_hours,
}
sampling_result = await ctx.sample(
f"Summarize this water level anomaly in 2 sentences for a "
f"marine operator:\n{json.dumps(anomaly_input)}",
system_prompt=(
"You are a marine conditions monitor. Be concise and factual. "
"State the risk level, deviation magnitude, and recommended action."
),
max_tokens=128,
)
if sampling_result.text:
result["summary"] = sampling_result.text
except Exception:
pass # Client doesn't support sampling
return result

21
src/mcnoaa_tides/tools/stations.py
View File

@ -1,12 +1,15 @@
"""Station discovery tools: search, proximity, and metadata lookup."""
from fastmcp import Context, FastMCP
from mcp.types import ToolAnnotations
from mcnoaa_tides.client import NOAAClient
_ANNOTATIONS = ToolAnnotations(readOnlyHint=True, openWorldHint=True)
def register(mcp: FastMCP) -> None:
@mcp.tool(tags={"discovery"})
@mcp.tool(tags={"discovery"}, annotations=_ANNOTATIONS)
async def search_stations(
ctx: Context,
query: str = "",
@ -24,9 +27,13 @@ def register(mcp: FastMCP) -> None:
"""
noaa: NOAAClient = ctx.lifespan_context["noaa_client"]
results = await noaa.search(query=query, state=state, is_tidal=is_tidal)
await ctx.debug(
f"search_stations: {len(results)} matches "
f"(query={query!r}, state={state!r}, is_tidal={is_tidal})"
)
return [s.model_dump() for s in results[:50]]
@mcp.tool(tags={"discovery"})
@mcp.tool(tags={"discovery"}, annotations=_ANNOTATIONS)
async def find_nearest_stations(
ctx: Context,
latitude: float,
@ -50,12 +57,17 @@ def register(mcp: FastMCP) -> None:
results = await noaa.find_nearest(
latitude, longitude, limit=limit, max_distance=max_distance_nm
)
return [
out = [
{**station.model_dump(), "distance_nm": round(dist, 1)}
for station, dist in results
]
await ctx.debug(
f"find_nearest_stations: {len(out)} stations within {max_distance_nm} nm "
f"of ({latitude}, {longitude})"
)
return out
@mcp.tool(tags={"discovery"})
@mcp.tool(tags={"discovery"}, annotations=_ANNOTATIONS)
async def get_station_info(
ctx: Context,
station_id: str,
@ -69,4 +81,5 @@ def register(mcp: FastMCP) -> None:
Example: get_station_info(station_id="8454000") for Providence, RI.
"""
noaa: NOAAClient = ctx.lifespan_context["noaa_client"]
await ctx.info(f"Fetching metadata for station {station_id}")
return await noaa.get_station_metadata(station_id)

19
src/mcnoaa_tides/tools/tides.py
View File

@ -1,12 +1,15 @@
"""Tide prediction and observed water level tools."""
from fastmcp import Context, FastMCP
from mcp.types import ToolAnnotations
from mcnoaa_tides.client import NOAAClient
_ANNOTATIONS = ToolAnnotations(readOnlyHint=True, openWorldHint=True)
def register(mcp: FastMCP) -> None:
@mcp.tool(tags={"tides"})
@mcp.tool(tags={"tides"}, annotations=_ANNOTATIONS)
async def get_tide_predictions(
ctx: Context,
station_id: str,
@ -30,6 +33,7 @@ def register(mcp: FastMCP) -> None:
t = timestamp, v = water level (ft), type = "H" or "L" (hilo only).
"""
noaa: NOAAClient = ctx.lifespan_context["noaa_client"]
await ctx.info(f"Fetching {interval} predictions for station {station_id}")
return await noaa.get_data(
station_id,
product="predictions",
@ -40,7 +44,7 @@ def register(mcp: FastMCP) -> None:
datum=datum,
)
@mcp.tool(tags={"tides"})
@mcp.tool(tags={"tides"}, annotations=_ANNOTATIONS)
async def get_observed_water_levels(
ctx: Context,
station_id: str,
@ -59,7 +63,8 @@ def register(mcp: FastMCP) -> None:
Quality flag "p" = preliminary, "v" = verified.
"""
noaa: NOAAClient = ctx.lifespan_context["noaa_client"]
return await noaa.get_data(
await ctx.info(f"Fetching observed water levels for station {station_id}")
result = await noaa.get_data(
station_id,
product="water_level",
begin_date=begin_date,
@ -67,3 +72,11 @@ def register(mcp: FastMCP) -> None:
hours=hours,
datum=datum,
)
# Flag preliminary data — important quality signal for downstream consumers
preliminary = [r for r in result.get("data", []) if r.get("q") == "p"]
if preliminary:
await ctx.warning(
f"Station {station_id}: {len(preliminary)} of "
f"{len(result.get('data', []))} readings are preliminary (unverified)"
)
return result

11
tests/conftest.py
View File

@ -10,7 +10,15 @@ from fastmcp.utilities.tests import run_server_async
from mcnoaa_tides import prompts, resources
from mcnoaa_tides.client import NOAAClient
from mcnoaa_tides.tools import charts, conditions, meteorological, smartpot, stations, tides
from mcnoaa_tides.tools import (
charts,
conditions,
diagnostics,
meteorological,
smartpot,
stations,
tides,
)
# Realistic station fixtures
MOCK_STATIONS_RAW = [
@ -155,6 +163,7 @@ def _build_test_server() -> FastMCP:
conditions.register(mcp)
smartpot.register(mcp)
charts.register(mcp)
diagnostics.register(mcp)
resources.register(mcp)
prompts.register(mcp)
return mcp

4
tests/test_charts.py
View File

@ -170,6 +170,6 @@ async def test_visualization_tools_registered(mcp_client: Client):
async def test_total_tool_count(mcp_client: Client):
"""Verify total tool count (9 base + 4 SmartPot = 13)."""
"""Verify total tool count (9 base + 4 SmartPot + 1 diagnostics = 14)."""
tools = await mcp_client.list_tools()
assert len(tools) == 13
assert len(tools) == 14

4
tests/test_smartpot.py
View File

@ -38,9 +38,9 @@ async def test_smartpot_tools_registered(mcp_client: Client):
async def test_total_tool_count(mcp_client: Client):
"""Should have 13 tools total (9 original + 4 SmartPot)."""
"""Should have 14 tools total (9 original + 4 SmartPot + 1 diagnostics)."""
tools = await mcp_client.list_tools()
assert len(tools) == 13
assert len(tools) == 14
# --- tidal_phase ---

13
tests/test_tools_stations.py
View File

@ -6,7 +6,7 @@ from fastmcp import Client
async def test_tool_registration(mcp_client: Client):
"""All 13 tools should be registered (9 original + 4 SmartPot)."""
"""All 14 tools should be registered (9 original + 4 SmartPot + 1 diagnostics)."""
tools = await mcp_client.list_tools()
tool_names = {t.name for t in tools}
expected = {
@ -23,10 +23,21 @@ async def test_tool_registration(mcp_client: Client):
"deployment_briefing",
"catch_tidal_context",
"water_level_anomaly",
"test_client_capabilities",
}
assert expected == tool_names
async def test_tool_annotations(mcp_client: Client):
"""All tools should declare readOnlyHint and openWorldHint annotations."""
tools = await mcp_client.list_tools()
for tool in tools:
assert tool.annotations is not None, f"{tool.name} missing annotations"
assert tool.annotations.readOnlyHint is True, (
f"{tool.name} should have readOnlyHint=True"
)
async def test_search_stations_by_state(mcp_client: Client):
result = await mcp_client.call_tool("search_stations", {"state": "RI"})
stations = json.loads(result.content[0].text)