Clari Production Checklist

Overview

Clari provides revenue intelligence through forecast data, pipeline analytics, and deal inspection. A production integration typically exports forecast snapshots, syncs pipeline data to a warehouse, and powers revenue dashboards. Incorrect data pipelines mean unreliable forecasts, missed quota signals, or stale deal intelligence that undermines board-level reporting.

Authentication & Secrets

• [ ] CLARIAPIKEY stored in secrets manager (not config files)
• [ ] Token tested against production endpoint before go-live
• [ ] Key rotation procedure documented (quarterly cycle)
• [ ] Separate tokens for dev/staging/prod environments
• [ ] Service account with least-privilege scopes (read-only for exports)

API Integration

• [ ] Production base URL configured (https://api.clari.com/v1)
• [ ] Rate limit handling with exponential backoff
• [ ] All required typesToExport configured (forecast, quota, crm_closed)
• [ ] Time period coverage verified (current quarter + 4 historical)
• [ ] Deduplication logic handles re-exports and overlapping periods
• [ ] Pagination implemented for large pipeline result sets
• [ ] Export job polling with configurable timeout (default: 10 min)

Error Handling & Resilience

• [ ] Circuit breaker configured for Clari API outages
• [ ] Retry with backoff for 429/5xx responses
• [ ] Empty export results handled (data quality alert, not silent pass)
• [ ] Export job timeout detection with automatic re-queue
• [ ] MERGE/UPSERT in warehouse prevents duplicate forecast records
• [ ] Data retention policy enforced (rolling 8 quarters typical)

Monitoring & Alerting

• [ ] API latency tracked per export job
• [ ] Error rate alerts set (threshold: any export failure)
• [ ] Forecast amount anomaly detection (>20% swing triggers review)
• [ ] Pipeline health dashboard with job completion rates
• [ ] Daily reconciliation: exported row counts vs expected

Validation Script

async function checkClariReadiness(): Promise<void> {
  const checks: { name: string; pass: boolean; detail: string }[] = [];
  // API connectivity
  try {
    const res = await fetch('https://api.clari.com/v1/forecast/types', {
      headers: { Authorization: `Bearer ${process.env.CLARI_API_KEY}` },
    });
    checks.push({ name: 'Clari API', pass: res.ok, detail: res.ok ? 'Connected' : `HTTP ${res.status}` });
  } catch (e: any) { checks.push({ name: 'Clari API', pass: false, detail: e.message }); }
  // Credentials present
  checks.push({ name: 'API Key Set', pass: !!process.env.CLARI_API_KEY, detail: process.env.CLARI_API_KEY ? 'Present

Clari Reference Architecture

Overview

Production architecture for Clari revenue intelligence integrations: export pipeline design, data warehouse schema, analytics layer, and alerting.

Architecture Diagram

┌──────────────┐     ┌─────────────────┐     ┌──────────────────┐
│  Clari App   │     │  Clari Export    │     │  Data Warehouse  │
│  (SaaS)      │────▶│  API (v4)       │────▶│  (Snowflake/BQ)  │
└──────────────┘     └─────────────────┘     └────────┬─────────┘
                                                       │
                     ┌─────────────────┐     ┌────────▼─────────┐
                     │  Change         │     │  Analytics /     │
                     │  Detection      │────▶│  Dashboard       │
                     └─────────────────┘     │  (Looker/Metabase)│
                            │                └──────────────────┘
                     ┌──────▼──────────┐
                     │  Alerts         │
                     │  (Slack/Email)  │
                     └─────────────────┘

Project Structure

clari-data-platform/
├── src/
│   ├── clari_client.py         # API client wrapper
│   ├── export_pipeline.py      # ETL pipeline
│   ├── change_detector.py      # Forecast change tracking
│   ├── models.py               # Data models
│   └── config.py               # Environment config
├── dags/
│   └── clari_export_dag.py     # Airflow DAG
├── sql/
│   ├── schema.sql              # Warehouse table definitions
│   ├── merge.sql               # Upsert logic
│   └── analytics/
│       ├── forecast_accuracy.sql
│       ├── pipeline_coverage.sql
│       └── rep_performance.sql
├── tests/
│   ├── fixtures/               # Sample API responses
│   ├── test_pipeline.py
│   └── test_change_detector.py
├── scripts/
│   ├── run_export.sh
│   └── validate_schema.py
└── monitoring/
    ├── alerts.yaml             # Alert rules
    └── dashboard.json          # Grafana/Looker config

Data Warehouse Schema

-- Core tables
CREATE TABLE clari_forecasts (
    id BIGINT GENERATED ALWAYS AS IDENTITY,
    owner_name VARCHAR NOT NULL,
    owner_email VARCHAR NOT NULL,
    forecast_amount DECIMAL(15,2),
    quota_amount DECIMAL(15,2),
    crm_total DECIMAL(15,2),
    crm_closed DECIMAL(15,2),
    adjustment_amount DECIMAL(15,2),
    time_period VARCHAR NOT NULL,
    forecast_name VARCHAR NOT NULL,
    exported_at TIMESTAMP NOT NULL,
    PRIMARY KEY (owner_email, time_period, forecast_name, exported_at)
);

-- Change tracking
CREATE TABLE clari_forecast_changes (
    id BIGINT GENERATED ALWAYS AS IDENTITY,
    owner_email VARCHAR NOT NULL,
    time_period VARCHAR NOT NULL,
    previous_amount DECIMAL(15,2),
    current_amount DECIMAL(15,2),
    change_pct DECIMAL(5,2),
    detected_at TIMESTAMP NOT NULL
);

-- Analytics views
CREATE VIEW v_forecast_accuracy AS
SELECT
    time_period,
    owner_name,
    forecast_amount,

Clari Upgrade & Migration

Overview

Handle Clari API changes: version migrations, export schema updates, and Copilot API adoption.

Instructions

Step 1: Check Current API Version

# v4 is the current version
curl -s -H "apikey: ${CLARI_API_KEY}" \
  https://api.clari.com/v4/export/forecast/list | jq .

# If using v3 (deprecated), migrate to v4

Step 2: Schema Change Detection

def detect_schema_changes(
    current_export: dict, expected_fields: set[str]
) -> dict:
    if not current_export.get("entries"):
        return {"status": "empty", "changes": []}

    actual_fields = set(current_export["entries"][0].keys())
    new_fields = actual_fields - expected_fields
    removed_fields = expected_fields - actual_fields

    return {
        "status": "changed" if new_fields or removed_fields else "compatible",
        "new_fields": list(new_fields),
        "removed_fields": list(removed_fields),
    }

# Track expected schema
EXPECTED_FIELDS = {
    "ownerName", "ownerEmail", "forecastAmount", "quotaAmount",
    "crmTotal", "crmClosed", "adjustmentAmount", "timePeriod"
}

Step 3: Database Schema Migration

-- Add new columns when Clari adds export fields
ALTER TABLE clari_forecasts ADD COLUMN IF NOT EXISTS new_field_name VARCHAR;

-- Backfill historical data
UPDATE clari_forecasts SET new_field_name = 'default' WHERE new_field_name IS NULL;

Rollback

Keep the previous client version alongside the new one until migration is verified:

# Pin client to specific behavior
client_v4 = ClariClient(ClariConfig(api_key=api_key, base_url="https://api.clari.com/v4"))

Resources

• Clari Developer Portal
• Clari Community

Next Steps

For CI integration, see clari-ci-integration.

Clari Webhooks & Events

Overview

Clari does not provide real-time webhooks. Instead, build change detection by comparing periodic exports. This skill covers scheduled export diffing, Slack alerts for forecast movements, and Copilot webhook integration.

Instructions

Step 1: Forecast Change Detection Pipeline

# forecast_monitor.py
import json
from pathlib import Path
from datetime import datetime

def detect_changes(
    current: list[dict],
    previous: list[dict],
    threshold_pct: float = 10.0,
) -> list[dict]:
    prev_map = {e["ownerEmail"]: e for e in previous}
    changes = []

    for entry in current:
        prev = prev_map.get(entry["ownerEmail"])
        if not prev:
            continue

        prev_fc = prev["forecastAmount"]
        curr_fc = entry["forecastAmount"]
        if prev_fc == 0:
            continue

        change_pct = ((curr_fc - prev_fc) / prev_fc) * 100
        if abs(change_pct) >= threshold_pct:
            changes.append({
                "rep": entry["ownerName"],
                "previous": prev_fc,
                "current": curr_fc,
                "change_pct": round(change_pct, 1),
                "direction": "increased" if change_pct > 0 else "decreased",
                "detected_at": datetime.utcnow().isoformat(),
            })

    return sorted(changes, key=lambda x: abs(x["change_pct"]), reverse=True)

def save_snapshot(entries: list[dict], path: str = "data/latest.json"):
    Path(path).parent.mkdir(exist_ok=True)
    with open(path, "w") as f:
        json.dump(entries, f)

def load_snapshot(path: str = "data/latest.json") -> list[dict]:
    try:
        with open(path) as f:
            return json.load(f)
    except FileNotFoundError:
        return []

Step 2: Slack Alert for Forecast Changes

import requests

def send_forecast_alert(changes: list[dict], slack_webhook: str):
    if not changes:
        return

    blocks = [f"*Clari Forecast Changes Detected*\n"]
    for c in changes[:10]:
        emoji = ":chart_with_upwards_trend:" if c["direction"] == "increased" else ":chart_with_downwards_trend:"
        blocks.append(
            f"{emoji} *{c['rep']}*: ${c['previous']:,.0f} -> ${c['current']:,.0f} "
            f"({c['change_pct']:+.1f}%)"
        )

    requests.post(slack_webhook, json={"text": "\n".join(blocks)})

Step 3: Scheduled Monitor (Cron)

#!/bin/bash
# Run every 4 hours: 0 */4 * * * /path/to/clari-monitor.sh
cd /opt/clari-integration
python3 -c "
from clari_client import ClariClient
from forecas