cmr_local_python

"""
agentarium_cmr_local.search — in-process Python wrapper around NASA's
open CMR REST API.

This is the canonical bundled-python reference tool for the Agentarium
registry. It is intentionally small: ~50 lines of code, one dependency
(requests, which the OpenAI Agents SDK pulls in transitively anyway).

Usage with OpenAI Agents SDK:

    from agents import Agent, Runner
    from agentarium_cmr_local.search import (
        search_collections, get_collection_metadata, get_granules,
    )

    agent = Agent(
        name="my-agent",
        instructions="...",
        tools=[search_collections, get_collection_metadata, get_granules],
    )

The SDK introspects each function's signature + type hints to construct
the JSON-schema the model sees. No MCP server, no Bearer token.
"""
from __future__ import annotations

from typing import Optional

import requests

_CMR_BASE = "https://cmr.earthdata.nasa.gov/search"
_TIMEOUT = 30  # CMR is usually fast; 30s lets slow queries complete.


def search_collections(
    keyword: str,
    temporal: Optional[str] = None,
    bounding_box: Optional[str] = None,
    provider: Optional[str] = None,
    page_size: int = 15,
    page_num: int = 1,
) -> dict:
    """Search CMR collections by keyword + optional temporal / spatial filters.

    Args:
        keyword: Free-text search (e.g. "sea ice concentration").
        temporal: ISO range "<start>,<end>" (e.g. "2023-01-01T00:00:00Z,2023-12-31T23:59:59Z").
        bounding_box: "west,south,east,north" in degrees. Arctic ≈ "-180,60,180,90".
        provider: CMR provider id (e.g. "NSIDC_ECS") to scope results.
        page_size: Results per page (CMR caps at 2000; default 15 for chat-sized output).
        page_num: 1-indexed page.

    Returns:
        {"hits": int, "items": [<collection-json>, ...]} — items follow CMR's UMM-C v1.x
        collection shape. Errors return {"error": "<message>", "status_code": int}.
    """
    params: dict[str, object] = {"keyword": keyword, "page_size": page_size, "page_num": page_num}
    if temporal:
        params["temporal"] = temporal
    if bounding_box:
        params["bounding_box"] = bounding_box
    if provider:
        params["provider"] = provider
    return _get_json("/collections.umm_json", params)


def get_collection_metadata(concept_id: str) -> dict:
    """Fetch full UMM-C metadata for one collection by concept ID (e.g. "C1234567890-NSIDC")."""
    return _get_json(f"/concepts/{concept_id}.umm_json", {})


def get_granules(
    concept_id: str,
    temporal: Optional[str] = None,
    bounding_box: Optional[str] = None,
    page_size: int = 10,
) -> dict:
    """List granules within a collection. Optional temporal / spatial filters."""
    params: dict[str, object] = {"collection_concept_id": concept_id, "page_size": page_size}
    if temporal:
        params["temporal"] = temporal
    if bounding_box:
        params["bounding_box"] = bounding_box
    return _get_json("/granules.umm_json", params)


def _get_json(path: str, params: dict) -> dict:
    """Internal: GET against CMR + return parsed JSON, or an {"error": ...} dict on failure.

    Network-level failures are converted to a structured error rather than
    raised, so the LLM tool-call loop can see + recover from them.
    """
    try:
        resp = requests.get(_CMR_BASE + path, params=params, timeout=_TIMEOUT)
    except requests.RequestException as exc:
        return {"error": f"network failure contacting CMR: {exc}", "status_code": None}
    if resp.status_code != 200:
        return {
            "error": f"CMR returned non-200 status",
            "status_code": resp.status_code,
            "body_excerpt": resp.text[:500],
        }
    return resp.json()