feat: custom metric definition CRUD + backup support

cortexapps_cli/commands/backup.py

@@ -220,9 +220,18 @@ def _export_catalog_pages(ctx, directory):
         except Exception as e:
             print(f"Failed to export catalog page {slug}: {e}")
 
+def _export_custom_metric_definitions(ctx, directory):
+    directory = _directory_name(directory, "custom-metric-definitions")
+    client = ctx.obj["client"]
+    data = client.fetch("api/v1/eng-intel/custom-metrics", params={})
+    for definition in data.get("customMetrics", []):
+        key = definition["key"]
+        _file_name(directory, key, definition, "json")
+
 backupTypes = sorted([
         "catalog",
         "catalogs",
+        "custom-metric-definitions",
         "entity-relationship-types",
         "entity-relationships",
         "entity-types",
@@ -273,6 +282,7 @@ def export(
     Exports the following objects:
     - catalog
     - catalogs
+    - custom-metric-definitions
     - entity-types
     - entity-relationship-types
     - entity-relationships
@@ -293,7 +303,7 @@ def export(
     - api-keys
     - custom-events
     - custom-metadata created by the public API
-    - custom-metrics
+    - custom-metrics data points (definitions are backed up; per-entity data points are not)
     - dependencies created by the API
     - deploys
     - docs created by the API
@@ -312,6 +322,10 @@ def export(
     _create_directory(directory)
     if "catalog" in export_types:
         _export_catalog(ctx, directory, catalog_types)
+    if "catalogs" in export_types:
+        _export_catalog_pages(ctx, directory)
+    if "custom-metric-definitions" in export_types:
+        _export_custom_metric_definitions(ctx, directory)
     if "entity-types" in export_types:
         _export_entity_types(ctx, directory)
     if "entity-relationship-types" in export_types:
@@ -326,8 +340,6 @@ def export(
         _export_scorecards(ctx, directory)
     if "workflows" in export_types:
         _export_workflows(ctx, directory)
-    if "catalogs" in export_types:
-        _export_catalog_pages(ctx, directory)
 
     print("\nExport complete!")
     print("Contents available in " + directory)
@@ -489,6 +501,33 @@ def import_relationships_file(file_info):
 
     return ("entity-relationships", len(results) - failed_count, [(fp, et, em) for rt, fp, et, em in results if et])
 
+def _import_custom_metric_definitions(ctx, directory):
+    imported = 0
+    failed = []
+    if os.path.isdir(directory):
+        print("Processing: " + directory)
+        client = ctx.obj["client"]
+        for filename in sorted(os.listdir(directory)):
+            if not filename.endswith(".json"):
+                continue
+            file_path = os.path.join(directory, filename)
+            if not os.path.isfile(file_path):
+                continue
+            try:
+                print("   Importing: " + filename)
+                with open(file_path) as f:
+                    definition = json.load(f)
+                client.post("api/v1/eng-intel/custom-metrics", data=definition)
+                imported += 1
+            except Exception as e:
+                if "409" in str(e):
+                    print(f"   Definition {filename} already exists, skipping")
+                    imported += 1
+                else:
+                    print(f"   Failed: {type(e).__name__} - {str(e)}")
+                    failed.append((file_path, type(e).__name__, str(e)))
+    return ("custom-metric-definitions", imported, failed)
+
 def _import_custom_metrics(ctx, directory):
     imported = 0
     failed = []
@@ -846,6 +885,10 @@ def import_tenant(
 ):
     """
     Import data into tenant
+
+    If the import directory contains a custom-metric-definitions/ subdirectory,
+    definitions are created before custom-metrics data points are imported.
+    Existing definitions (HTTP 409) are skipped silently and counted as successes.
     """
 
     client = ctx.obj["client"]
@@ -857,6 +900,7 @@ def import_tenant(
     all_stats.append(_import_entity_relationship_types(ctx, directory + "/entity-relationship-types"))
     all_stats.append(_import_catalog(ctx, directory + "/catalog"))
     all_stats.append(_import_entity_relationships(ctx, directory + "/entity-relationships"))
+    all_stats.append(_import_custom_metric_definitions(ctx, directory + "/custom-metric-definitions"))
     all_stats.append(_import_custom_metrics(ctx, directory + "/custom-metrics"))
     all_stats.append(_import_plugins(ctx, directory + "/plugins"))
     all_stats.append(_import_scorecards(ctx, directory + "/scorecards"))
@@ -910,6 +954,8 @@ def import_tenant(
             elif import_type == "entity-relationships":
                 # These need special handling - would need the relationship type
                 print(f"# Manual retry needed for entity-relationships: {file_path}")
+            elif import_type == "custom-metric-definitions":
+                print(f"cortex custom-metrics definition-create -f \"{file_path}\"")
             elif import_type == "custom-metrics":
                 print(f"# Manual retry needed for custom-metrics: {file_path}")
             elif import_type == "plugins":

cortexapps_cli/commands/custom_metrics.py

@@ -1,7 +1,10 @@
 from collections import defaultdict
 from datetime import datetime
+import json
 import typer
+from typing import Optional
 from typing_extensions import Annotated
+from rich import print_json
 
 app = typer.Typer(help="Custom metrics commands")
 
@@ -131,3 +134,109 @@ def delete(
     params = _convert_datetime_to_string(params)
 
     r = client.delete("api/v1/eng-intel/custom-metrics/" + custom_metric_key + "/entity/" + tag, params=params)
+
+@app.command()
+def definition_create(
+    ctx: typer.Context,
+    file_input: Annotated[typer.FileText, typer.Option("--file", "-f", help="JSON file containing the metric definition; can be passed as stdin with -, example: -f-")] = None,
+    key: Optional[str] = typer.Option(None, "--key", "-k", help="Unique metric key (2-255 chars, letters/digits/hyphens, alphanumeric start/end)"),
+    name: Optional[str] = typer.Option(None, "--name", "-n", help="Display name for the metric"),
+    valence: Optional[str] = typer.Option(None, "--valence", "-V", help="Whether higher values are preferable: POSITIVE, NEGATIVE, or NEUTRAL"),
+    category: Optional[str] = typer.Option(None, "--category", "-c", help="Grouping label"),
+    description: Optional[str] = typer.Option(None, "--description", "-d", help="What the metric measures"),
+    is_draft: bool = typer.Option(False, "--is-draft", help="If true, metric accepts data but is hidden from Eng Intelligence"),
+):
+    """
+    Create a custom metric definition
+    """
+
+    client = ctx.obj["client"]
+
+    if file_input:
+        data = json.loads("".join([line for line in file_input]))
+    else:
+        if not key or not name or not valence:
+            raise typer.BadParameter("--key, --name, and --valence are required when --file is not provided")
+        data = {
+            "key": key,
+            "name": name,
+            "valence": valence,
+            "category": category,
+            "description": description,
+            "isDraft": is_draft,
+            "type": "PUBLIC_API"
+        }
+        # Remove keys whose value is None
+        data = {k: v for k, v in data.items() if v is not None}
+
+    result = client.post("api/v1/eng-intel/custom-metrics", data=data)
+    print_json(data=result)
+
+@app.command()
+def definition_list(
+    ctx: typer.Context,
+    page: int | None = typer.Option(None, "--page", "-p", help="Page number, 0-indexed — omit to fetch all pages"),
+    page_size: int | None = typer.Option(None, "--page-size", "-z", help="Page size for results"),
+    prt: bool = typer.Option(True, "--print", help="If result should be printed to the terminal", hidden=True),
+):
+    """List custom metric definitions"""
+
+    client = ctx.obj["client"]
+
+    params = {}
+    if page_size is not None:
+        params["pageSize"] = page_size
+
+    client.fetch_or_get("api/v1/eng-intel/custom-metrics", page, prt, params=params)
+
+@app.command()
+def definition_get(
+    ctx: typer.Context,
+    key: str = typer.Option(..., "--key", "-k", help="Metric key"),
+):
+    """Get a custom metric definition"""
+
+    client = ctx.obj["client"]
+
+    result = client.get("api/v1/eng-intel/custom-metrics/" + key)
+    print_json(data=result)
+
+@app.command()
+def definition_update(
+    ctx: typer.Context,
+    key: str = typer.Option(..., "--key", "-k", help="Metric key to update"),
+    file_input: Annotated[typer.FileText, typer.Option("--file", "-f", help="JSON file with updated definition; can be passed as stdin with -, example: -f-")] = None,
+    name: Optional[str] = typer.Option(None, "--name", "-n", help="Display name for the metric"),
+    valence: Optional[str] = typer.Option(None, "--valence", "-V", help="POSITIVE, NEGATIVE, or NEUTRAL"),
+    category: Optional[str] = typer.Option(None, "--category", "-c", help="Grouping label"),
+    description: Optional[str] = typer.Option(None, "--description", "-d", help="What the metric measures"),
+):
+    """Update a custom metric definition"""
+
+    client = ctx.obj["client"]
+
+    if file_input:
+        data = json.loads("".join([line for line in file_input]))
+    else:
+        data = {
+            "name": name,
+            "valence": valence,
+            "category": category,
+            "description": description,
+        }
+        # Strip None-valued keys
+        data = {k: v for k, v in data.items() if v is not None}
+
+    result = client.put("api/v1/eng-intel/custom-metrics/" + key, data=data)
+    print_json(data=result)
+
+@app.command()
+def definition_delete(
+    ctx: typer.Context,
+    key: str = typer.Option(..., "--key", "-k", help="Metric key to delete"),
+):
+    """Delete a custom metric definition"""
+
+    client = ctx.obj["client"]
+
+    client.delete("api/v1/eng-intel/custom-metrics/" + key)

cortexapps_cli/solutions/ai-spend/README.md

@@ -65,11 +65,6 @@ Register every employee as a Cortex entity linked to their team, push weekly Cla
 | Sync script | `scripts/sync-claude-spend.py` |
 | GH Actions workflow | `.github/workflows/sync-claude-spend.yaml` |
 
-## Prerequisites
-
-Before installing, create the `ai-spend` custom metric definition in your Cortex instance:
-**Eng Intel → Custom Metrics → New Metric**, key: `ai-spend`.
-
 ## Installation
 
 ```
@@ -161,5 +156,5 @@ The sample entities are fictional. Add your real employees as catalog entities w
 
 - Users who authenticate Claude Code with a personal API key (not Enterprise OAuth) show $0 spend in the Analytics API and are skipped automatically.
 - Cost data may take up to 24 hours to appear; dates at least 30 days old are considered final for billing purposes.
-- The `ai-spend` custom metric definition must currently be created manually before installing. A future release will support auto-creation of custom metric definitions as part of `cortex solutions install`.
+- The `ai-spend` custom metric definition is installed automatically via `custom-metric-definitions/ai-spend.json` when you run `cortex solutions install -s ai-spend`.
 - The scorecard's Bronze rule uses a `P1Y` lookback to accommodate sample data. Once your weekly sync is running consistently, consider tightening it to `P8D` to ensure the rule only passes when data is fresh. The Silver and Gold rules use `P8D` and can similarly be adjusted to match your sync frequency.

cortexapps_cli/solutions/ai-spend/custom-metric-definitions/ai-spend.json

@@ -0,0 +1,9 @@
+{
+  "key": "ai-spend",
+  "name": "AI Spend",
+  "type": "PUBLIC_API",
+  "valence": "POSITIVE",
+  "category": "Cost",
+  "description": "Weekly Claude AI spend per entity in USD",
+  "isDraft": false
+}

tests/test_backup.py

@@ -56,6 +56,49 @@ def test_backup_import_invalid_api_key(monkeypatch):
         assert result.exit_code != 0, f"backup import should exit with non-zero code on failure, got exit_code={result.exit_code}"
 
 
+def test_backup_import_custom_metric_definitions_invalid_api_key(monkeypatch):
+    """
+    Test that backup import of custom-metric-definitions fails cleanly with invalid API key.
+    """
+    monkeypatch.setenv("CORTEX_API_KEY", "invalidKey")
+
+    with tempfile.TemporaryDirectory() as tmpdir:
+        defs_dir = os.path.join(tmpdir, "custom-metric-definitions")
+        os.makedirs(defs_dir)
+
+        def_file = os.path.join(defs_dir, "test-metric.json")
+        with open(def_file, "w") as f:
+            json.dump({
+                "key": "test-metric",
+                "name": "Test Metric",
+                "valence": "POSITIVE",
+                "type": "PUBLIC_API",
+                "isDraft": True,
+            }, f)
+
+        result = cli(["backup", "import", "-d", tmpdir], return_type=ReturnType.RAW)
+        assert result.exit_code != 0, (
+            f"backup import should exit with non-zero code on failure, "
+            f"got exit_code={result.exit_code}"
+        )
+
+
+def test_backup_import_custom_metric_definitions_empty_dir():
+    """
+    Test that backup import with an empty custom-metric-definitions/ dir is a silent no-op.
+    """
+    with tempfile.TemporaryDirectory() as tmpdir:
+        defs_dir = os.path.join(tmpdir, "custom-metric-definitions")
+        os.makedirs(defs_dir)
+        # No files — import should succeed with 0 imported and 0 failed
+        result = cli(["backup", "import", "-d", tmpdir], return_type=ReturnType.RAW)
+        # Empty dir is a no-op — exit code 0 (all_stats has 0 failed)
+        assert result.exit_code == 0, (
+            f"backup import with empty custom-metric-definitions/ should exit 0, "
+            f"got exit_code={result.exit_code}\n{result.stdout}"
+        )
+
+
 def test_backup_export_invalid_api_key(monkeypatch):
     """
     Test that backup export exits with non-zero return code and clean error message when API calls fail.