From db6ed2f550161c461085e2b5aabffb1c9973bad9 Mon Sep 17 00:00:00 2001 From: Ryan Malloy Date: Wed, 18 Feb 2026 18:30:32 -0700 Subject: [PATCH] Use FastMCP Image type, Context progress, and dynamic resources MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Replace hand-rolled base64 dicts with FastMCP's Image() helper in get_imagery, compare_dates, get_imagery_composite, and get_legend. FastMCP handles base64 encoding and MIME type automatically. Add Context progress reporting to imagery tools so MCP clients can display fetch status (resolving location, fetching imagery, etc). Add dynamic resources: gibs://colormap/{layer_id} — colormap explanation text gibs://dates/{layer_id} — date range + live DescribeDomains Remove unused base64 import. --- src/mcgibs/server.py | 106 +++++++++++++++++++++-------------- tests/test_tools.py | 128 +++++++++++++++++++++++++++++++++++++++++++ 2 files changed, 193 insertions(+), 41 deletions(-) diff --git a/src/mcgibs/server.py b/src/mcgibs/server.py index 29101f1..a5bf7dc 100644 --- a/src/mcgibs/server.py +++ b/src/mcgibs/server.py @@ -10,13 +10,14 @@ FastMCP needs runtime types for Literal, list[str], etc. import asyncio import atexit -import base64 import json import logging import httpx from fastmcp import FastMCP +from fastmcp.server.context import Context from fastmcp.server.middleware import Middleware +from fastmcp.utilities.types import Image from mcgibs.capabilities import search_layers from mcgibs.client import GIBSClient @@ -306,12 +307,13 @@ async def _resolve_bbox( async def get_imagery( layer_id: str, date: str, + ctx: Context, bbox: list[float] | None = None, place: str | None = None, width: int = 1024, height: int = 1024, format: str = "jpeg", -) -> list[dict]: +): """Fetch GIBS imagery via WMS. Args: @@ -327,22 +329,22 @@ async def get_imagery( layer = client.get_layer(layer_id) if layer is None: - return [{"type": "text", "text": f"Layer '{layer_id}' not found."}] + return f"Layer '{layer_id}' not found." + await ctx.report_progress(1, 4, "Resolving location...") try: resolved_bbox = await _resolve_bbox(client, bbox, place) except Exception as exc: - return [{"type": "text", "text": str(exc)}] - - image_format = f"image/{format}" + return str(exc) + await ctx.report_progress(2, 4, "Fetching imagery from GIBS...") image_bytes = await client.get_wms_image( layer_id, date, resolved_bbox, width, height, - image_format, + f"image/{format}", ) description = ( @@ -350,18 +352,11 @@ async def get_imagery( f"Region: {place or resolved_bbox.wms_bbox}\n" f"Size: {width}x{height}" ) - - # If the layer has a colormap, add a hint about explain_colormap if layer.has_colormap: - description += "\nTip: use explain_colormap to understand what the colors represent." + description += "\nTip: use explain_layer_colormap to understand what the colors represent." - mime = f"image/{format}" - b64 = base64.b64encode(image_bytes).decode() - - return [ - {"type": "text", "text": description}, - {"type": "image", "data": b64, "mimeType": mime}, - ] + await ctx.report_progress(4, 4, "Complete") + return [description, Image(data=image_bytes, format=format)] @mcp.tool( @@ -372,9 +367,10 @@ async def compare_dates( layer_id: str, date_before: str, date_after: str, + ctx: Context, bbox: list[float] | None = None, place: str | None = None, -) -> list[dict]: +): """Side-by-side comparison of two dates. Args: @@ -388,13 +384,16 @@ async def compare_dates( layer = client.get_layer(layer_id) if layer is None: - return [{"type": "text", "text": f"Layer '{layer_id}' not found."}] + return f"Layer '{layer_id}' not found." + await ctx.report_progress(1, 5, "Resolving location...") try: resolved_bbox = await _resolve_bbox(client, bbox, place) except Exception as exc: - return [{"type": "text", "text": str(exc)}] + return str(exc) + await ctx.report_progress(2, 5, f"Fetching imagery for {date_before}...") + await ctx.report_progress(3, 5, f"Fetching imagery for {date_after}...") composite_bytes = await client.compare_dates( layer_id, date_before, @@ -408,11 +407,8 @@ async def compare_dates( f"Region: {place or resolved_bbox.wms_bbox}" ) - b64 = base64.b64encode(composite_bytes).decode() - return [ - {"type": "text", "text": description}, - {"type": "image", "data": b64, "mimeType": "image/jpeg"}, - ] + await ctx.report_progress(5, 5, "Complete") + return [description, Image(data=composite_bytes, format="jpeg")] @mcp.tool( @@ -422,11 +418,12 @@ async def compare_dates( async def get_imagery_composite( layer_ids: list[str], date: str, + ctx: Context, bbox: list[float] | None = None, place: str | None = None, width: int = 1024, height: int = 1024, -) -> list[dict]: +): """Multi-layer composite image. Args: @@ -440,13 +437,16 @@ async def get_imagery_composite( client = _get_client() if len(layer_ids) > 5: - return [{"type": "text", "text": "WMS supports at most 5 layers per composite."}] + return "WMS supports at most 5 layers per composite." + await ctx.report_progress(1, 3, "Resolving location...") try: resolved_bbox = await _resolve_bbox(client, bbox, place) except Exception as exc: - return [{"type": "text", "text": str(exc)}] + return str(exc) + layer_names = ", ".join(layer_ids) + await ctx.report_progress(2, 3, f"Fetching {len(layer_ids)}-layer composite...") image_bytes = await client.get_wms_composite( layer_ids, date, @@ -455,16 +455,12 @@ async def get_imagery_composite( height, ) - layer_names = ", ".join(layer_ids) description = ( f"Composite: {layer_names}\nDate: {date}\nRegion: {place or resolved_bbox.wms_bbox}" ) - b64 = base64.b64encode(image_bytes).decode() - return [ - {"type": "text", "text": description}, - {"type": "image", "data": b64, "mimeType": "image/jpeg"}, - ] + await ctx.report_progress(3, 3, "Complete") + return [description, Image(data=image_bytes, format="jpeg")] # ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━ @@ -490,7 +486,7 @@ async def explain_layer_colormap(layer_id: str) -> str: async def get_legend( layer_id: str, orientation: str = "horizontal", -) -> list[dict]: +): """Fetch the legend graphic for a layer. Args: @@ -501,13 +497,9 @@ async def get_legend( legend_bytes = await client.get_legend_image(layer_id, orientation) if legend_bytes is None: - return [{"type": "text", "text": f"No legend available for '{layer_id}'."}] + return f"No legend available for '{layer_id}'." - b64 = base64.b64encode(legend_bytes).decode() - return [ - {"type": "text", "text": f"Legend for {layer_id}"}, - {"type": "image", "data": b64, "mimeType": "image/png"}, - ] + return [f"Legend for {layer_id}", Image(data=legend_bytes, format="png")] # ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━ @@ -631,6 +623,38 @@ async def layer_resource(layer_id: str) -> str: return layer.model_dump_json(indent=2) +@mcp.resource("gibs://colormap/{layer_id}") +async def colormap_resource(layer_id: str) -> str: + """Natural-language explanation of a layer's colormap.""" + client = _get_client() + return await client.explain_layer_colormap(layer_id) + + +@mcp.resource("gibs://dates/{layer_id}") +async def dates_resource(layer_id: str) -> str: + """Available date range for a layer.""" + client = _get_client() + layer = client.get_layer(layer_id) + if layer is None: + return json.dumps({"error": f"Layer '{layer_id}' not found"}) + + info: dict = {} + if layer.time: + info["start"] = layer.time.start + info["end"] = layer.time.end + info["period"] = layer.time.period + info["default"] = layer.time.default + + try: + domains = await client.describe_domains(layer_id) + if "time_domain" in domains: + info["live_time_domain"] = domains["time_domain"] + except (httpx.HTTPError, RuntimeError): + pass + + return json.dumps(info, indent=2) + + @mcp.resource("gibs://projections") async def projections_resource() -> str: """Available GIBS projections with endpoint information.""" diff --git a/tests/test_tools.py b/tests/test_tools.py index 5a85436..ba4e4d7 100644 --- a/tests/test_tools.py +++ b/tests/test_tools.py @@ -1,10 +1,12 @@ """FastMCP integration tests — tool calls against the real server, HTTP mocked.""" import json +from io import BytesIO import httpx import respx from fastmcp import Client +from PIL import Image as PILImage import mcgibs.server as server_module from mcgibs.client import GIBSClient @@ -171,3 +173,129 @@ async def test_list_tools(capabilities_xml): finally: await server_module._client.close() server_module._client = None + + +# --------------------------------------------------------------------------- +# Image() return type tests +# --------------------------------------------------------------------------- + + +def _make_fake_jpeg() -> bytes: + """Create a tiny valid JPEG for mocking WMS responses.""" + buf = BytesIO() + PILImage.new("RGB", (10, 10), "blue").save(buf, format="JPEG") + return buf.getvalue() + + +@respx.mock +async def test_get_imagery_returns_image_content(capabilities_xml): + """get_imagery returns TextContent + ImageContent (not raw dicts).""" + respx.get(url__regex=r".*WMTSCapabilities\.xml").mock( + return_value=httpx.Response(200, text=capabilities_xml) + ) + respx.get(url__regex=r".*wms\.cgi.*").mock( + return_value=httpx.Response( + 200, content=_make_fake_jpeg(), headers={"content-type": "image/jpeg"} + ) + ) + + server_module._client = await _init_mock_client(capabilities_xml) + try: + async with Client(mcp) as client: + result = await client.call_tool( + "get_imagery", + { + "layer_id": "MODIS_Terra_CorrectedReflectance_TrueColor", + "date": "2025-06-01", + "bbox": [-120.0, 30.0, -110.0, 40.0], + }, + ) + assert len(result.content) == 2 + assert result.content[0].type == "text" + assert "Corrected Reflectance" in result.content[0].text + assert result.content[1].type == "image" + assert result.content[1].mimeType == "image/jpeg" + finally: + await server_module._client.close() + server_module._client = None + + +@respx.mock +async def test_get_legend_returns_image_content(capabilities_xml): + """get_legend returns TextContent + ImageContent for layers with legends.""" + respx.get(url__regex=r".*WMTSCapabilities\.xml").mock( + return_value=httpx.Response(200, text=capabilities_xml) + ) + + fake_png = BytesIO() + PILImage.new("RGB", (200, 20), "white").save(fake_png, format="PNG") + respx.get(url__regex=r".*legends/.*\.png").mock( + return_value=httpx.Response( + 200, content=fake_png.getvalue(), headers={"content-type": "image/png"} + ) + ) + + server_module._client = await _init_mock_client(capabilities_xml) + try: + async with Client(mcp) as client: + result = await client.call_tool( + "get_legend", + {"layer_id": "AMSR2_Sea_Ice_Concentration_12km_Monthly"}, + ) + assert len(result.content) == 2 + assert result.content[0].type == "text" + assert result.content[1].type == "image" + assert result.content[1].mimeType == "image/png" + finally: + await server_module._client.close() + server_module._client = None + + +# --------------------------------------------------------------------------- +# Dynamic resource tests +# --------------------------------------------------------------------------- + + +@respx.mock +async def test_list_resources(capabilities_xml): + """Server exposes all expected resources including dynamic templates.""" + respx.get(url__regex=r".*WMTSCapabilities\.xml").mock( + return_value=httpx.Response(200, text=capabilities_xml) + ) + + server_module._client = await _init_mock_client(capabilities_xml) + try: + async with Client(mcp) as client: + templates = await client.list_resource_templates() + template_uris = {t.uriTemplate for t in templates} + + assert "gibs://layer/{layer_id}" in template_uris + assert "gibs://colormap/{layer_id}" in template_uris + assert "gibs://dates/{layer_id}" in template_uris + finally: + await server_module._client.close() + server_module._client = None + + +@respx.mock +async def test_colormap_resource(capabilities_xml, colormap_xml): + """gibs://colormap/{layer_id} returns colormap explanation text.""" + respx.get(url__regex=r".*WMTSCapabilities\.xml").mock( + return_value=httpx.Response(200, text=capabilities_xml) + ) + respx.get(url__regex=r".*colormaps/v1\.3/.*\.xml").mock( + return_value=httpx.Response(200, text=colormap_xml) + ) + + server_module._client = await _init_mock_client(capabilities_xml) + try: + async with Client(mcp) as client: + result = await client.read_resource( + "gibs://colormap/AIRS_L3_Surface_Air_Temperature_Daily_Day" + ) + text = result[0].text + assert "Surface Air Temperature" in text + assert "K" in text + finally: + await server_module._client.close() + server_module._client = None