Add cds toolset: Copernicus Climate Data Store retrieval workflow

pyproject.toml

@@ -10,6 +10,7 @@ dependencies = [
     "mcp-agent",
     "dataset-search",
     "aoi-generator",
+    "cds",
 ]
 
 [dependency-groups]
@@ -32,6 +33,7 @@ mcp-cli = { workspace = true }
 mcp-agent = { workspace = true }
 dataset-search = { workspace = true }
 aoi-generator = { workspace = true }
+cds = { workspace = true }
 
 [tool.ruff]
 line-length = 88

toolsets/cds/pyproject.toml

@@ -0,0 +1,19 @@
+[project]
+name = "cds"
+version = "0.1.0"
+description = "Copernicus Climate Data Store (CDS) tools: search datasets, inspect schemas, submit and track retrieval jobs."
+requires-python = ">=3.12,<3.14"
+dependencies = [
+    "langchain-core<2.0.0,>=1.4.6",
+    "mcp-runtime",
+    "httpx>=0.27",
+    "pydantic-settings>=2.5",
+    "tenacity>=9.0",
+]
+
+[tool.uv.sources]
+mcp-runtime = { workspace = true }
+
+[build-system]
+requires = ["hatchling"]
+build-backend = "hatchling.build"

toolsets/cds/src/cds/__init__.py

@@ -0,0 +1 @@
+"""cds toolset: Copernicus Climate Data Store tools, ported from cds-assistant."""

toolsets/cds/src/cds/client.py

@@ -0,0 +1,14 @@
+import httpx
+
+from .settings import settings
+
+
+def make_client() -> httpx.AsyncClient:
+    return httpx.AsyncClient(
+        base_url=settings.cds_api_url,
+        headers={
+            "PRIVATE-TOKEN": settings.cds_api_key,
+            "Content-Type": "application/json",
+        },
+        timeout=httpx.Timeout(connect=10.0, read=120.0, write=10.0, pool=10.0),
+    )

toolsets/cds/src/cds/settings.py

@@ -0,0 +1,21 @@
+from pydantic_settings import BaseSettings, SettingsConfigDict
+
+
+class Settings(BaseSettings):
+    """CDS API configuration, read from the environment (or a .env in the cwd).
+
+    `cds_api_key` defaults to empty so the toolset imports and serves without
+    credentials; calls then return structured auth errors until CDS_API_KEY
+    is provided (in k8s, via the Secret listed in toolset.yaml).
+    """
+
+    model_config = SettingsConfigDict(
+        env_file=".env", env_file_encoding="utf-8", extra="ignore"
+    )
+
+    cds_api_key: str = ""
+    cds_api_url: str = "https://cds.climate.copernicus.eu/api/retrieve/v1"
+    cds_catalogue_url: str = "https://cds.climate.copernicus.eu/api/catalogue/v1"
+
+
+settings = Settings()

toolsets/cds/src/cds/tools/__init__.py

@@ -0,0 +1,38 @@
+"""LangChain tools for the Copernicus Climate Data Store (CDS).
+
+Covers the full retrieval workflow: search the catalogue, inspect a
+dataset's schema and constraints, submit requests, then poll jobs and
+fetch download links.
+"""
+
+from .apply_constraints import apply_constraints
+from .check_credentials import check_credentials
+from .get_dataset_schema import get_dataset_schema
+from .get_job_status import get_job_status
+from .get_results import get_results
+from .list_jobs import list_jobs
+from .search_datasets import search_datasets
+from .submit_request import submit_request
+
+__all__ = [
+    "TOOLS",
+    "apply_constraints",
+    "check_credentials",
+    "get_dataset_schema",
+    "get_job_status",
+    "get_results",
+    "list_jobs",
+    "search_datasets",
+    "submit_request",
+]
+
+TOOLS = [
+    search_datasets,
+    get_dataset_schema,
+    apply_constraints,
+    submit_request,
+    get_job_status,
+    get_results,
+    list_jobs,
+    check_credentials,
+]

toolsets/cds/src/cds/tools/_client.py

@@ -0,0 +1,12 @@
+import httpx
+
+from ..client import make_client
+
+_client: httpx.AsyncClient | None = None
+
+
+def get_client() -> httpx.AsyncClient:
+    global _client
+    if _client is None:
+        _client = make_client()
+    return _client

toolsets/cds/src/cds/tools/_errors.py

@@ -0,0 +1,79 @@
+from typing import Any
+
+import httpx
+
+
+def auth_error(detail: str = "Invalid or missing API key.") -> dict[str, Any]:
+    return {"error": "auth", "detail": detail}
+
+
+def bad_request_error(detail: str) -> dict[str, Any]:
+    return {"error": "bad_request", "detail": detail}
+
+
+def licence_error(dataset: str = "") -> dict[str, Any]:
+    if dataset:
+        url = f"https://cds.climate.copernicus.eu/datasets/{dataset}"
+        detail = f"Dataset licence not accepted. Accept the terms at {url} then retry."
+    else:
+        detail = (
+            "Dataset licence not accepted. "
+            "Visit the dataset page on CDS and accept the terms before retrying."
+        )
+    return {"error": "licence", "detail": detail}
+
+
+def queue_limit_error() -> dict[str, Any]:
+    return {
+        "error": "queue_limit",
+        "detail": (
+            "Concurrent queued-job limit reached. Wait for existing jobs to complete."
+        ),
+    }
+
+
+def not_found_error(detail: str) -> dict[str, Any]:
+    return {"error": "not_found", "detail": detail}
+
+
+def not_ready_error(job_id: str, status: str) -> dict[str, Any]:
+    return {
+        "error": "not_ready",
+        "detail": (
+            f"Job {job_id!r} is not yet successful (status: {status!r}). "
+            "Poll again later."
+        ),
+    }
+
+
+def transient_error(detail: str) -> dict[str, Any]:
+    return {"error": "transient", "detail": detail}
+
+
+def classify_submit_failure(message: str, dataset: str = "") -> dict[str, Any]:
+    """Map a failed-job traceback/message to the appropriate error code."""
+    lower = message.lower()
+    if "licence" in lower or "license" in lower or "terms" in lower:
+        return licence_error(dataset)
+    if "queue" in lower:
+        return queue_limit_error()
+    return bad_request_error(message)
+
+
+def classify_http_error(resp: httpx.Response, dataset: str = "") -> dict[str, Any]:
+    if resp.status_code == 401:
+        return auth_error()
+    try:
+        body: Any = resp.json()
+        detail: Any = body.get("detail", resp.text)
+    except Exception:
+        body = {}
+        detail = resp.text
+    if resp.status_code == 403:
+        detail_str = str(detail).lower()
+        if any(w in detail_str for w in ("licence", "license", "terms")):
+            return licence_error(dataset)
+        return auth_error(str(detail))
+    if resp.status_code in (400, 422):
+        return bad_request_error(str(detail))
+    return transient_error(f"HTTP {resp.status_code}: {str(detail)[:200]}")

toolsets/cds/src/cds/tools/_retry.py

@@ -0,0 +1,33 @@
+from collections.abc import Callable
+from typing import Any, TypeVar
+
+import httpx
+from tenacity import (
+    retry,
+    retry_if_exception_type,
+    stop_after_attempt,
+    wait_exponential,
+)
+
+# Exceptions that warrant automatic backoff retry.
+# 5xx errors must be raised explicitly via raise_for_status() to trigger this.
+TRANSIENT_EXC = (
+    httpx.TimeoutException,
+    httpx.ConnectError,
+    httpx.RemoteProtocolError,
+    httpx.HTTPStatusError,
+)
+
+_F = TypeVar("_F", bound=Callable[..., Any])
+
+_retry = retry(
+    retry=retry_if_exception_type(TRANSIENT_EXC),
+    stop=stop_after_attempt(3),
+    wait=wait_exponential(multiplier=1, min=2, max=10),
+    reraise=True,
+)
+
+
+def with_retry(fn: _F) -> _F:
+    """Typed wrapper around tenacity retry so decorated functions keep their signature."""
+    return _retry(fn)

toolsets/cds/src/cds/tools/apply_constraints.py

@@ -0,0 +1,50 @@
+from typing import Any
+
+from langchain_core.tools import tool
+
+from ._client import get_client
+from ._errors import classify_http_error, not_found_error, transient_error
+from ._retry import TRANSIENT_EXC, with_retry
+
+
+@with_retry
+async def _call(dataset: str, partial_request: dict[str, Any]) -> dict[str, Any]:
+    resp = await get_client().post(
+        f"/processes/{dataset}/constraints",
+        json={"inputs": partial_request},
+    )
+    if resp.status_code >= 500:
+        resp.raise_for_status()
+    if resp.status_code == 404:
+        return not_found_error(f"Dataset {dataset!r} not found.")
+    if resp.status_code != 200:
+        return classify_http_error(resp)
+
+    raw: dict[str, list[str]] = resp.json()
+    # Omit parameters with empty lists — those are unconstrained (e.g. area)
+    valid_values = {k: v for k, v in raw.items() if v}
+
+    return {"dataset": dataset, "valid_values": valid_values}
+
+
+@tool
+async def apply_constraints(
+    dataset: str, partial_request: dict[str, Any]
+) -> dict[str, Any]:
+    """Get valid parameter values for a CDS dataset given a partial request.
+
+    Pass whatever parameters you already know (e.g. year and month) and receive
+    back the valid values for all remaining parameters. Use this to verify that
+    requested dates exist in the archive — especially for recent or current-year data.
+
+    Args:
+        dataset: Dataset identifier, e.g. "reanalysis-era5-land".
+        partial_request: Partial request dict, e.g. {"year": "2026", "month": "05"}.
+
+    Returns a dict with 'dataset' and 'valid_values' mapping each constrained
+    parameter to its list of currently valid values.
+    """
+    try:
+        return await _call(dataset, partial_request)
+    except TRANSIENT_EXC as exc:
+        return transient_error(str(exc))

toolsets/cds/src/cds/tools/check_credentials.py

@@ -0,0 +1,26 @@
+from typing import Any
+
+from langchain_core.tools import tool
+
+from ._client import get_client
+from ._errors import classify_http_error, transient_error
+from ._retry import TRANSIENT_EXC, with_retry
+
+
+@with_retry
+async def _call() -> dict[str, Any]:
+    resp = await get_client().get("/jobs", params={"limit": 1})
+    if resp.status_code >= 500:
+        resp.raise_for_status()
+    if resp.status_code == 200:
+        return {"ok": True}
+    return classify_http_error(resp)
+
+
+@tool
+async def check_credentials() -> dict[str, Any]:
+    """Validate the CDS API key. Returns {"ok": True} on success or a structured error dict."""
+    try:
+        return await _call()
+    except TRANSIENT_EXC as exc:
+        return transient_error(str(exc))