STAC MCP Server

Instructions

Implementation Reference

• stac_mcp/tools/estimate_data_size.py:85-265 (handler)

  Core handler function that validates input arguments, calls STACClient.estimate_data_size, and formats the result as text or JSON.

  def handle_estimate_data_size(
      client: STACClient,
      arguments: dict[str, Any],
  ) -> list[TextContent] | dict[str, Any]:
      collections = _validate_collections_argument(arguments.get("collections"))
      bbox = _validate_bbox_argument(arguments.get("bbox"))
      dt = _validate_datetime_argument(arguments.get("datetime"))
      query = _validate_query_argument(arguments.get("query"))
      aoi_geojson = _validate_aoi_geojson_argument(arguments.get("aoi_geojson"))
      limit = arguments.get("limit", 10)
      force_metadata_only = arguments.get("force_metadata_only", False)
  
      size_estimate = client.estimate_data_size(
          collections=collections,
          bbox=bbox,
          datetime=dt,
          query=query,
          aoi_geojson=aoi_geojson,
          limit=limit,
          force_metadata_only=force_metadata_only,
      )
      # Note: we do not return JSON here immediately because we want to ensure
      # sensor-native and queried totals are computed and included in the JSON
      # output. The JSON branch is evaluated after computing MB/GB fallbacks.
      result_text = "**Data Size Estimation**\n\n"
      item_count = size_estimate.get("item_count", 0)
      result_text += f"Items analyzed: {item_count}\n"
  
      # Be defensive: some estimator implementations may omit the
      # pre-computed MB/GB fields. Prefer explicit fields but fall back to
      # reconstructing from bytes when necessary.
      estimated_bytes = size_estimate.get("estimated_size_bytes")
      if estimated_bytes is None:
          # Some older/test fixtures may use 'estimated_bytes' or 'estimated_size'
          estimated_bytes = size_estimate.get("estimated_bytes")
  
      estimated_mb = size_estimate.get("estimated_size_mb")
      if estimated_mb is None and estimated_bytes is not None:
          try:
              estimated_mb = float(estimated_bytes) / (1024 * 1024)
          except (TypeError, ValueError):
              estimated_mb = None
  
      estimated_gb = size_estimate.get("estimated_size_gb")
      if estimated_gb is None and estimated_mb is not None:
          try:
              estimated_gb = float(estimated_mb) / 1024.0
          except (TypeError, ValueError):
              estimated_gb = None
  
      est_mb_str = (
          f"{estimated_mb:.2f} MB" if isinstance(estimated_mb, (int, float)) else "n/a"
      )
      est_gb_str = (
          f"{estimated_gb:.4f} GB" if isinstance(estimated_gb, (int, float)) else "n/a"
      )
      result_text += f"Estimated size: {est_mb_str} ({est_gb_str})\n"
  
      # Always surface sensor-native totals to the agent and the user.
      # Some estimator implementations compute an instrument-native (sensor) total
      # for narrower dtype suggestions; expose those values explicitly here.
      sensor_bytes = size_estimate.get("sensor_native_estimated_size_bytes")
      if sensor_bytes is None:
          sensor_bytes = size_estimate.get("sensor_native_estimated_bytes")
  
      sensor_mb = size_estimate.get("sensor_native_estimated_size_mb")
      if sensor_mb is None and sensor_bytes is not None:
          try:
              sensor_mb = float(sensor_bytes) / (1024 * 1024)
          except (TypeError, ValueError):
              sensor_mb = None
  
      sensor_gb = size_estimate.get("sensor_native_estimated_size_gb")
      if sensor_gb is None and sensor_mb is not None:
          try:
              sensor_gb = float(sensor_mb) / 1024.0
          except (TypeError, ValueError):
              sensor_gb = None
  
      s_mb_str = f"{sensor_mb:.2f} MB" if isinstance(sensor_mb, (int, float)) else "n/a"
      s_gb_str = f"{sensor_gb:.4f} GB" if isinstance(sensor_gb, (int, float)) else "n/a"
      result_text += f"Sensor-native estimated size: {s_mb_str} ({s_gb_str})\n"
      raw_bytes_str = (
          f"{int(estimated_bytes):,}" if estimated_bytes is not None else "n/a"
      )
      result_text += f"Raw bytes: {raw_bytes_str}\n\n"
      result_text += "**Query Parameters:**\n"
      result_text += "Collections: "
      collections_list = (
          ", ".join(size_estimate["collections"])
          if size_estimate["collections"]
          else "All"
      )
      result_text += f"{collections_list}\n"
      if size_estimate["bbox_used"]:
          b = size_estimate["bbox_used"]
          result_text += (
              f"Bounding box: [{b[0]:.4f}, {b[1]:.4f}, {b[2]:.4f}, {b[3]:.4f}]\n"
          )
      if size_estimate["temporal_extent"]:
          result_text += f"Time range: {size_estimate['temporal_extent']}\n"
      if size_estimate["clipped_to_aoi"]:
          result_text += "Clipped to AOI: Yes (minimized to smallest area)\n"
      if "data_variables" in size_estimate:
          result_text += "\n**Data Variables:**\n"
          for var_info in size_estimate["data_variables"]:
              # Support multiple possible size keys produced by different
              # estimator implementations/tests: prefer explicit 'size_mb',
              # then 'estimated_size_mb', then compute from 'estimated_bytes'.
              size_mb = None
              if "size_mb" in var_info:
                  size_mb = var_info["size_mb"]
              elif "estimated_size_mb" in var_info:
                  size_mb = var_info["estimated_size_mb"]
              elif (
                  "estimated_bytes" in var_info
                  and var_info["estimated_bytes"] is not None
              ):
                  try:
                      size_mb = var_info["estimated_bytes"] / (1024 * 1024)
                  except (TypeError, ValueError):
                      size_mb = None
  
              size_str = f"{size_mb:.2f}" if isinstance(size_mb, (int, float)) else "n/a"
              result_text += (
                  f"  - {var_info.get('variable', 'unknown')}: {size_str} MB, "
                  f"shape {var_info.get('shape')}, dtype {var_info.get('dtype')}\n"
              )
      if size_estimate.get("spatial_dims"):
          spatial = size_estimate["spatial_dims"]
          result_text += "\n**Spatial Dimensions:**\n"
          result_text += f"  X (longitude): {spatial.get('x', 0)} pixels\n"
          result_text += f"  Y (latitude): {spatial.get('y', 0)} pixels\n"
      if "assets_analyzed" in size_estimate:
          result_text += "\n**Assets Analyzed (fallback estimation):**\n"
          for asset_info in size_estimate["assets_analyzed"][:MAX_ASSET_LIST]:
              result_text += (
                  f"  - {asset_info['asset']}: {asset_info['estimated_size_mb']} MB "
                  f"({asset_info['media_type']})\n"
              )
          remaining = len(size_estimate["assets_analyzed"]) - MAX_ASSET_LIST
          if remaining > 0:
              result_text += f"  ... and {remaining} more assets\n"
      result_text += f"\n{size_estimate['message']}\n"
      # If JSON was requested, return a structured payload that includes both
      # the queried totals and the sensor-native totals so agents can rely on
      # a stable schema.
      if arguments.get("output_format") == "json":
          queried_totals = {
              "bytes": estimated_bytes,
              "mb": estimated_mb,
              "gb": estimated_gb,
          }
          sensor_native_totals = {
              "bytes": sensor_bytes,
              "mb": sensor_mb,
              "gb": sensor_gb,
          }
          return {
              "type": "data_size_estimate",
              "estimate": size_estimate,
              "queried_totals": queried_totals,
              "sensor_native_totals": sensor_native_totals,
          }
  
      # Append advisory guidance from the dtype prompt if available. This helps
      # agents and human users understand how to prefer compact dtypes and avoid
      # overestimation when NaN nodata forces float upcasts.
      if callable(dtype_size_preferences):
          try:
              advisory = dtype_size_preferences()
              if advisory:
                  result_text += "\n**Estimator Advisory (dtype preferences)**\n"
                  result_text += advisory + "\n"
          except (
              RuntimeError,
              TypeError,
              ValueError,
          ) as exc:  # pragma: no cover - best-effort
              _LOGGER.debug("estimate_data_size: advisory generation failed: %s", exc)
      return [TextContent(type="text", text=result_text)]

• stac_mcp/server.py:112-143 (registration)

  MCP tool registration with @app.tool decorator. Preprocesses arguments and delegates to execution.execute_tool("estimate_data_size"). The type hints define the input schema.

  @app.tool
  async def estimate_data_size(
      collections: list[str] | str,
      bbox: list[float] | str | None = None,
      datetime: str | None = None,
      query: dict[str, Any] | str | None = None,
      aoi_geojson: dict[str, Any] | str | None = None,
      limit: int | None = 10,
      force_metadata_only: bool | None = False,
      output_format: str | None = "text",
      catalog_url: str | None = None,
  ) -> list[dict[str, Any]]:
      """Estimate the data size for a STAC query."""
      arguments = preprocess_parameters(
          {
              "collections": collections,
              "bbox": bbox,
              "datetime": datetime,
              "query": query,
              "aoi_geojson": aoi_geojson,
              "limit": limit,
              "force_metadata_only": force_metadata_only,
              "output_format": output_format,
          }
      )
      return await execution.execute_tool(
          "estimate_data_size",
          arguments=arguments,
          catalog_url=catalog_url,
          headers=None,
      )

• stac_mcp/tools/execution.py:56-67 (registration)

  Tool handler registry mapping "estimate_data_size" to handle_estimate_data_size function.

  _TOOL_HANDLERS: dict[str, Handler] = {
      "search_collections": handle_search_collections,
      "get_collection": handle_get_collection,
      "search_items": handle_search_items,
      "get_item": handle_get_item,
      "estimate_data_size": handle_estimate_data_size,
      "get_root": handle_get_root,
      "get_conformance": handle_get_conformance,
      "get_queryables": handle_get_queryables,
      "get_aggregations": handle_get_aggregations,
      "sensor_registry_info": handle_sensor_registry_info,
  }

• stac_mcp/tools/estimate_data_size.py:32-83 (schema)

  Input validation functions used by the handler to enforce schema for parameters like collections, bbox, datetime, query, aoi_geojson.

  def _validate_collections_argument(
      collections: list[str] | None,
  ) -> list[str]:
      match collections:
          case None:
              msg = "Collections argument is required."
              raise ValueError(msg)
          case []:
              msg = "Collections argument cannot be empty."
              raise ValueError(msg)
          case _:
              return collections
  
  
  def _validate_datetime_argument(dt: str | None) -> str | None:
      """Datetime may be omitted. If 'latest' is provided, return today's date string."""
      if dt is None or dt == "":
          return None
      if dt == "latest":
          return f"{get_today_date()}"
      return dt
  
  
  def _validate_query_argument(query: dict[str, Any] | None) -> dict[str, Any] | None:
      """Query is optional for estimate; return as-is (may be None)."""
      return query
  
  
  def _validate_bbox_argument(bbox: list[float] | None) -> list[float] | None:
      """Validate bbox argument.
  
      BBox is optional for many STAC queries; if omitted, return None. If
      provided, it must be a sequence of four floats [minx, miny, maxx, maxy].
      """
      if bbox is None:
          return None
      bbox_len = 4
      # Accept any sequence of length 4
      if isinstance(bbox, (list, tuple)) and len(bbox) == bbox_len:
          return list(bbox)
      msg = (
          "Invalid bbox argument; must be a list of four floats: [minx, miny, maxx, maxy]"
      )
      raise ValueError(msg)
  
  
  def _validate_aoi_geojson_argument(
      aoi_geojson: dict[str, Any] | None,
  ) -> dict[str, Any] | None:
      """AOI GeoJSON is optional; return as-is (may be None)."""
      return aoi_geojson

• stac_mcp/tools/params.py:10-100 (helper)

  Parameter preprocessing utility called by server tool to convert string-serialized params (common in MCP) to native types.

  def preprocess_parameters(arguments: dict[str, Any]) -> dict[str, Any]:
      """Preprocess tool parameters to handle various input formats.
  
      This function normalizes parameters that may come in as strings but should be
      other types (arrays, objects, etc.). This is particularly useful when MCP clients
      serialize parameters as strings.
  
      Args:
          arguments: Raw arguments dictionary from MCP client
  
      Returns:
          Preprocessed arguments with proper types
      """
      if not arguments:
          return arguments
  
      processed = arguments.copy()
  
      # Handle bbox parameter - should be a list of 4 floats
      if "bbox" in processed and processed["bbox"] is not None:
          bbox = processed["bbox"]
          if isinstance(bbox, str):
              try:
                  # Try to parse as JSON
                  parsed = json.loads(bbox)
                  if isinstance(parsed, list) and len(parsed) == 4:  # noqa: PLR2004
                      processed["bbox"] = [float(x) for x in parsed]
                      logger.debug(
                          "Converted bbox from string to list: %s", processed["bbox"]
                      )
              except (json.JSONDecodeError, ValueError, TypeError) as e:
                  logger.warning("Failed to parse bbox string: %s, error: %s", bbox, e)
  
      # Handle collections parameter - should be a list of strings
      if "collections" in processed and processed["collections"] is not None:
          collections = processed["collections"]
          if isinstance(collections, str):
              try:
                  parsed = json.loads(collections)
                  if isinstance(parsed, list):
                      processed["collections"] = parsed
                      logger.debug(
                          "Converted collections from string to list: %s",
                          processed["collections"],
                      )
              except (json.JSONDecodeError, ValueError, TypeError) as e:
                  logger.warning(
                      "Failed to parse collections string: %s, error: %s", collections, e
                  )
  
      # Handle aoi_geojson parameter - should be a dict/object
      if "aoi_geojson" in processed and processed["aoi_geojson"] is not None:
          aoi = processed["aoi_geojson"]
          if isinstance(aoi, str):
              try:
                  parsed = json.loads(aoi)
                  if isinstance(parsed, dict):
                      processed["aoi_geojson"] = parsed
                      logger.debug("Converted aoi_geojson from string to dict")
              except (json.JSONDecodeError, ValueError, TypeError) as e:
                  logger.warning(
                      "Failed to parse aoi_geojson string: %s, error: %s", aoi, e
                  )
  
      # Handle query parameter - should be a dict/object
      if "query" in processed and processed["query"] is not None:
          query = processed["query"]
          if isinstance(query, str):
              try:
                  parsed = json.loads(query)
                  if isinstance(parsed, dict):
                      processed["query"] = parsed
                      logger.debug("Converted query from string to dict")
              except (json.JSONDecodeError, ValueError, TypeError) as e:
                  logger.warning("Failed to parse query string: %s, error: %s", query, e)
  
      if "limit" in processed and processed["limit"] is not None:
          limit = processed["limit"]
          if isinstance(limit, str):
              try:
                  processed["limit"] = int(limit)
                  logger.debug(
                      "Converted limit from string to int: %d", processed["limit"]
                  )
              except ValueError as e:
                  logger.warning(
                      "Failed to convert limit string to int: %s, error: %s", limit, e
                  )
  
      return processed