STAC MCP Server

Instructions

Implementation Reference

• stac_mcp/tools/search_items.py:12-54 (handler)

  Core handler function that executes the search_items tool logic: searches STAC items via STACClient, formats results as formatted text or JSON.
  def handle_search_items( client: STACClient, arguments: dict[str, Any], ) -> list[TextContent] | dict[str, Any]: collections = arguments.get("collections") bbox = arguments.get("bbox") dt = arguments.get("datetime") query = arguments.get("query") limit = arguments.get("limit", 10) items = client.search_items( collections=collections, bbox=bbox, datetime=dt, query=query, limit=limit, ) if arguments.get("output_format") == "json": return {"type": "item_list", "count": len(items), "items": items} result_text = f"Found {len(items)} items:\n\n" asset_keys = set() for item in items: item_id = item.get("id", "unknown") collection_id = item.get("collection", "unknown") result_text += f"**{item_id}** (Collection: `{collection_id}`)\n" dt_value = item.get("datetime") if dt_value: result_text += f" Date: {dt_value}\n" bbox = item.get("bbox") if isinstance(bbox, list | tuple) and len(bbox) >= BBOX_MIN_COORDS: result_text += ( " BBox: " f"[{bbox[0]:.2f}, {bbox[1]:.2f}, {bbox[2]:.2f}, {bbox[3]:.2f}]\n" ) assets = item.get("assets") or {} asset_keys.update(assets.keys()) asset_count = len(assets) if hasattr(assets, "__len__") else 0 result_text += f" Assets: {asset_count}\n\n" result_text += "\n" if asset_keys: result_text += "Assets found across items:\n" for key in sorted(asset_keys): result_text += f" - {key}\n" return [TextContent(type="text", text=result_text)]
• stac_mcp/server.py:83-109 (registration)

  MCP tool registration using @app.tool decorator, defines input parameters (schema) and dispatches to execution engine.
  @app.tool async def search_items( collections: list[str] | str, bbox: list[float] | str | None = None, datetime: str | None = None, limit: int | None = 10, query: dict[str, Any] | str | None = None, output_format: str | None = "text", catalog_url: str | None = None, ) -> list[dict[str, Any]]: """Search for STAC items.""" arguments = preprocess_parameters( { "collections": collections, "bbox": bbox, "datetime": datetime, "limit": limit, "query": query, "output_format": output_format, } ) return await execution.execute_tool( "search_items", arguments=arguments, catalog_url=catalog_url, headers=None, )
• stac_mcp/tools/execution.py:56-67 (registration)

  Internal registry mapping 'search_items' tool name to its handler function for dispatch in execute_tool.
  _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/params.py:10-100 (helper)

  Helper function to preprocess input parameters (e.g., parse stringified JSON for bbox, collections, query), used by search_items registration.
  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