Clari CI Integration

Overview

Add Clari export validation to CI: test API connectivity, validate export schemas, and run pipeline integration tests.

Prerequisites

• Protected CI environment with scoped, masked secrets
• Mock fixtures for pull-request tests and an approved integration test account
• A repository environment gate for any production-targeted job
• Retention policy for generated logs and artifacts

Instructions

GitHub Actions Workflow

name: Clari Pipeline Tests

on:
  push:
    paths: ["src/clari/**", "tests/clari/**"]
  schedule:
    - cron: "0 6 * * 1"  # Weekly Monday check

jobs:
  test:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - uses: actions/setup-python@v5
        with:
          python-version: "3.11"

      - run: pip install -r requirements.txt

      - name: Unit tests (mock data)
        run: pytest tests/ -v -k "not integration"

      - name: Integration test (real API)
        if: github.ref == 'refs/heads/main'
        env:
          CLARI_API_KEY: ${{ secrets.CLARI_API_KEY }}
        run: |
          python -c "
          from clari_client import ClariClient
          client = ClariClient()
          forecasts = client.list_forecasts()
          assert len(forecasts) > 0, 'No forecasts found'
          print(f'Connected: {len(forecasts)} forecasts available')
          "

      - name: Schema validation
        env:
          CLARI_API_KEY: ${{ secrets.CLARI_API_KEY }}
        run: |
          python scripts/validate_schema.py

Store Secrets

gh secret set CLARI_API_KEY --body "your-api-token"

Error Handling

Output

Publish a redacted run summary with fixture/integration status, schema version, external job IDs, artifact retention location, and promotion decision. Secrets, forecast payloads, and download URLs must remain masked and unavailable to untrusted pull-request code.

Examples

Run mock tests on every pull request, then run one read-only integration export only from a protected branch using a scoped environment secret. Fail closed if the schema cha…

Clari Common Errors

Overview

Diagnostic guide for the most common Clari API issues: authentication failures, empty exports, job timeouts, and data discrepancies.

Prerequisites

• Authorized, scoped access to the affected Clari environment
• Redacted job IDs, timestamps, and export metadata for diagnosis
• A named owner for credentials, forecast data, and production change approval
• Access to the certified prior dataset for safe comparison and recovery

Instructions

Confirm the environment and affected period, collect the smallest redacted evidence needed to classify the issue, then follow the matching reference entry. Change one variable at a time and do not retry authentication, rate, or data-integrity failures indefinitely. Escalate ambiguous provider behavior with job IDs and timestamps rather than guessing at a production fix.

Error Handling

Error Reference

1. 401 Unauthorized

{"error": "Unauthorized", "message": "Invalid API key"}

Fix: Regenerate token at Clari > User Settings > API Token. Tokens may expire or be revoked by admins.

2. 403 Forbidden -- API Access Not Enabled

{"error": "Forbidden", "message": "API access not enabled for this user"}

Fix: Contact your Clari admin to enable API access. Requires enterprise plan.

3. 404 Forecast Not Found

{"error": "Not Found", "message": "Forecast 'wrong_name' not found"}

Fix: List available forecasts first:

curl -s -H "apikey: ${CLARI_API_KEY}" \
  https://api.clari.com/v4/export/forecast/list | jq '.forecasts[].forecastName'

4. Export Returns Empty Entries

The API returns {"entries": []} with no error.

Causes:

• Time period has no submitted forecasts
• User lacks visibility into the forecast hierarchy
• Wrong forecast name (case-sensitive)

Fix: Verify in Clari UI that the forecast has submissions for the requested period.

Clari Core Workflow: Forecast Export Pipeline

Overview

Primary workflow: build an automated pipeline that exports forecast submissions, quota, adjustments, and CRM data from Clari to your data warehouse. Supports Snowflake, BigQuery, and PostgreSQL as targets.

Prerequisites

• Completed clari-install-auth and clari-sdk-patterns setup
• Target database or data warehouse with write access
• Python 3.10+ with requests and your DB driver

Instructions

Step 1: Define Export Configuration

# config.py
from dataclasses import dataclass

@dataclass
class ExportConfig:
    forecast_name: str          # From Clari forecast list
    time_periods: list[str]     # e.g., ["2026_Q1", "2025_Q4"]
    export_types: list[str] = None
    currency: str = "USD"
    include_historical: bool = True

    def __post_init__(self):
        if self.export_types is None:
            self.export_types = [
                "forecast",           # Submitted forecast call
                "forecast_updated",   # Updated forecast history
                "quota",              # Quota values
                "adjustment",         # Manager adjustments
                "crm_total",          # Total CRM pipeline
                "crm_closed",         # Closed-won CRM amounts
            ]

Step 2: Build the Export Pipeline

# export_pipeline.py
from clari_client import ClariClient
from config import ExportConfig
import json
from datetime import datetime

def run_export(config: ExportConfig) -> list[dict]:
    client = ClariClient()
    all_entries = []

    for period in config.time_periods:
        print(f"Exporting {config.forecast_name} for {period}...")

        data = client.export_and_download(
            forecast_name=config.forecast_name,
            time_period=period,
        )

        entries = data.get("entries", [])
        for entry in entries:
            entry["_exported_at"] = datetime.utcnow().isoformat()
            entry["_forecast_name"] = config.forecast_name

        all_entries.extend(entries)
        print(f"  {len(entries)} records exported")

    return all_entries

def transform_forecast_data(entries: list[dict]) -> dict:
    total_forecast = sum(e.get("forecastAmount", 0) for e in entries)
    total_quota = sum(e.get("quotaAmount", 0) for e in entries)
    total_closed = sum(e.get("crmClosed", 0) for e in entries)

    return {
        "total_forecast": total_forecast,
        "total_quota": total_quota,
        "total_closed": total_closed,
        "attainment_percent": (total_closed / total_quota * 100) if total_quota else 0,
      …

Clari Core Workflow: Revenue Analytics

Overview

Build revenue analytics from Clari export data: forecast accuracy tracking, pipeline coverage analysis, rep performance dashboards, and forecast call change detection.

Prerequisites

• Completed clari-core-workflow-a (export pipeline)
• Historical forecast exports for accuracy tracking
• Pandas/SQL for data analysis

Instructions

Step 1: Forecast Accuracy Analysis

import pandas as pd

def calculate_forecast_accuracy(
    forecasts: list[dict], actuals: list[dict]
) -> pd.DataFrame:
    df_forecast = pd.DataFrame(forecasts)
    df_actual = pd.DataFrame(actuals)

    merged = df_forecast.merge(
        df_actual[["ownerEmail", "crmClosed"]],
        on="ownerEmail",
        suffixes=("_forecast", "_actual"),
    )

    merged["accuracy_pct"] = (
        1 - abs(merged["forecastAmount"] - merged["crmClosed_actual"])
        / merged["forecastAmount"]
    ) * 100

    merged["variance"] = merged["crmClosed_actual"] - merged["forecastAmount"]

    return merged[["ownerName", "forecastAmount", "crmClosed_actual",
                    "accuracy_pct", "variance"]].sort_values("accuracy_pct")

Step 2: Pipeline Coverage Report

def pipeline_coverage_report(entries: list[dict]) -> dict:
    df = pd.DataFrame(entries)

    return {
        "total_pipeline": df["crmTotal"].sum(),
        "total_closed": df["crmClosed"].sum(),
        "total_quota": df["quotaAmount"].sum(),
        "total_forecast": df["forecastAmount"].sum(),
        "coverage_ratio": df["crmTotal"].sum() / df["quotaAmount"].sum()
            if df["quotaAmount"].sum() > 0 else 0,
        "close_rate": df["crmClosed"].sum() / df["crmTotal"].sum()
            if df["crmTotal"].sum() > 0 else 0,
        "attainment_pct": df["crmClosed"].sum() / df["quotaAmount"].sum() * 100
            if df["quotaAmount"].sum() > 0 else 0,
        "at_risk_reps": len(df[df["forecastAmount"] < df["quotaAmount"] * 0.7]),
        "on_track_reps": len(df[df["forecastAmount"] >= df["quotaAmount"] * 0.9]),
    }

Step 3: Forecast Change Detection

def detect_forecast_changes(
    current: list[dict], previous: list[dict], threshold_pct: float = 10.0
) -> list[dict]:
    curr = {e["ownerEmail"]: e for e in current}
    prev = {e["ownerEmail"]: e for e in pr…

Clari Cost Tuning

Overview

Minimize Clari API overhead: reduce export frequency, cache aggressively, export only needed data types, and monitor usage.

Prerequisites

• An approved consumer inventory and freshness requirement for each report
• Usage telemetry for API calls, export jobs, warehouse loads, and cache age
• Data-owner approval for retention, aggregation, and delivery changes
• A non-production validation path for schedule or payload changes

Instructions

Export Only What You Need

# Full export (6 data types) -- more API load
full_types = ["forecast", "quota", "forecast_updated",
              "adjustment", "crm_total", "crm_closed"]

# Minimal export (2 data types) -- faster and lighter
minimal_types = ["forecast", "crm_closed"]

# Use minimal for dashboards, full for audit/compliance

Optimize Export Frequency

Cache to Avoid Redundant Exports

# Cache recent exports (see clari-performance-tuning)
cache = ExportCache(ttl_hours=8)

def smart_export(client, forecast_name, period):
    cached = cache.get(forecast_name, period)
    if cached:
        print(f"Cache hit for {period}")
        return cached

    data = client.export_and_download(forecast_name, period)
    entries = data.get("entries", [])
    cache.set(forecast_name, period, entries)
    return entries

Usage Tracking

class ClariUsageTracker:
    def __init__(self):
        self.api_calls = 0
        self.exports = 0

    def track_call(self):
        self.api_calls += 1

    def track_export(self):
        self.exports += 1

    def report(self) -> dict:
        return {
            "api_calls": self.api_calls,
            "exports": self.exports,
        }

Error Handling

Clari Debug Bundle

Overview

Collect Clari API diagnostic information for support: API connectivity, forecast list, job history, and error responses. All secrets are redacted.

Prerequisites

• Authorized diagnostic access to the affected environment
• CLARI_API_KEY injected by an approved secret mechanism
• A secure local workspace with enough storage for the bundle
• A support/incident record that defines what metadata may be shared

Instructions

Debug Bundle Script

#!/bin/bash
# clari-debug-bundle.sh
set -euo pipefail

BUNDLE_DIR="clari-debug-$(date +%Y%m%d-%H%M%S)"
mkdir -p "$BUNDLE_DIR"

echo "=== Clari Debug Bundle ===" | tee "$BUNDLE_DIR/summary.txt"
echo "Generated: $(date -u)" | tee -a "$BUNDLE_DIR/summary.txt"

# 1. API connectivity
echo "--- API Connectivity ---" >> "$BUNDLE_DIR/summary.txt"
HTTP_CODE=$(curl -s -o /dev/null -w "%{http_code}" \
  -H "apikey: ${CLARI_API_KEY}" \
  https://api.clari.com/v4/export/forecast/list)
echo "API Status: HTTP ${HTTP_CODE}" >> "$BUNDLE_DIR/summary.txt"

# 2. Forecast list (no sensitive data)
curl -s -H "apikey: ${CLARI_API_KEY}" \
  https://api.clari.com/v4/export/forecast/list \
  | jq '.forecasts[] | {forecastName, forecastId}' \
  > "$BUNDLE_DIR/forecasts.json" 2>&1

# 3. Recent export jobs
curl -s -H "apikey: ${CLARI_API_KEY}" \
  https://api.clari.com/v4/export/jobs \
  | jq '.jobs[] | {jobId, status, createdAt, forecastName}' \
  > "$BUNDLE_DIR/jobs.json" 2>&1

# 4. Environment info (redacted)
echo "--- Environment ---" >> "$BUNDLE_DIR/summary.txt"
echo "CLARI_API_KEY: ${CLARI_API_KEY:+[SET]}" >> "$BUNDLE_DIR/summary.txt"
python3 --version >> "$BUNDLE_DIR/summary.txt" 2>&1
pip3 show requests 2>/dev/null | grep Version >> "$BUNDLE_DIR/summary.txt" || true

# 5. Package
tar -czf "$BUNDLE_DIR.tar.gz" "$BUNDLE_DIR"
rm -rf "$BUNDLE_DIR"
echo "Bundle: $BUNDLE_DIR.tar.gz"

Safe to share: Forecast names, job IDs, HTTP status codes, library versions. Never share: API key, forecast amounts, rep names, email addresses.

Error Handling

Clari Deploy Integration

Overview

Deploy Clari export pipelines to production environments: Airflow DAGs, AWS Lambda, or Google Cloud Functions for scheduled, serverless execution.

Prerequisites

• Environment-specific cloud project/account and least-privilege runtime role
• Secret-manager reference for the Clari token, never a literal deployment value
• A reviewed export schedule, retry policy, and dead-letter/incident route
• Separate non-production validation and production approval gates

Instructions

Airflow DAG

# dags/clari_export_dag.py
from airflow import DAG
from airflow.operators.python import PythonOperator
from airflow.models import Variable
from datetime import datetime, timedelta

def export_clari_forecast(**context):
    from clari_client import ClariClient, ClariConfig

    client = ClariClient(ClariConfig(
        api_key=Variable.get("clari_api_key"),
    ))

    period = context["params"].get("period", "2026_Q1")
    data = client.export_and_download("company_forecast", period)

    entries = data.get("entries", [])
    context["ti"].xcom_push(key="entry_count", value=len(entries))
    # Load to warehouse here

dag = DAG(
    "clari_daily_export",
    schedule_interval="0 6 * * *",
    start_date=datetime(2026, 1, 1),
    catchup=False,
    default_args={"retries": 2, "retry_delay": timedelta(minutes=5)},
)

export_task = PythonOperator(
    task_id="export_forecast",
    python_callable=export_clari_forecast,
    dag=dag,
)

AWS Lambda

# lambda_handler.py
import json
import boto3
from clari_client import ClariClient, ClariConfig

def handler(event, context):
    ssm = boto3.client("ssm")
    api_key = ssm.get_parameter(
        Name="/clari/api-key", WithDecryption=True
    )["Parameter"]["Value"]

    client = ClariClient(ClariConfig(api_key=api_key))
    data = client.export_and_download(
        event.get("forecast_name", "company_forecast"),
        event.get("period", "2026_Q1"),
    )

    return {
        "statusCode": 200,
        "body": json.dumps({"entries": len(data.get("entries", []))}),
    }

Google Cloud Function

# main.py
import functions_framework
from google.cloud import secretmanager
from clari_client import ClariClient, ClariConfig

@functions_framework.http
def clari_export(request):
    sm = secretmanager.SecretManagerServiceClient()
    secret = sm.access_secret_version(name="projects/my-proj/secrets/clari-api-key/versions/latest")
    api_key = secret.payload.data.decode()

    client = ClariClient(ClariConfig(a…

Clari Hello World

Overview

First API calls against Clari: list available forecasts, export a forecast snapshot, and check export job status. The Clari Export API is the primary integration point for getting forecast, quota, and CRM data out of Clari.

Prerequisites

• Completed clari-install-auth setup
• CLARI_API_KEY environment variable set
• At least one forecast configured in Clari

Instructions

Step 1: List Available Forecasts

curl -s -H "apikey: ${CLARI_API_KEY}" \
  https://api.clari.com/v4/export/forecast/list \
  | jq '.forecasts[] | {forecastName, forecastId, timePeriods}'

Step 2: Export a Forecast

import requests
import json
import os
import time

api_key = os.environ["CLARI_API_KEY"]
headers = {"apikey": api_key, "Content-Type": "text/plain"}

# Replace with your forecast name from Step 1
forecast_name = "company_forecast"

payload = json.dumps({
    "timePeriod": "2026_Q1",
    "typesToExport": [
        "forecast",
        "quota",
        "forecast_updated",
        "adjustment",
        "crm_total",
        "crm_closed"
    ],
    "currency": "USD",
    "schedule": "NONE",
    "includeHistorical": False,
    "exportFormat": "JSON"
})

response = requests.post(
    f"https://api.clari.com/v4/export/forecast/{forecast_name}",
    headers=headers,
    data=payload,
)
response.raise_for_status()

job = response.json()
print(f"Export job started: {job['jobId']}")
print(f"Status: {job['status']}")

Step 3: Check Export Job Status

# Poll for job completion
job_id = job["jobId"]

while True:
    status_resp = requests.get(
        f"https://api.clari.com/v4/export/jobs/{job_id}",
        headers={"apikey": api_key},
    )
    status = status_resp.json()

    if status["status"] == "COMPLETED":
        print(f"Export ready: {status['downloadUrl']}")
        break
    elif status["status"] == "FAILED":
        print(f"Export failed: {status.get('error', 'Unknown')}")
        break

    print(f"Status: {status['status']}... waiting 5s")
    time.sleep(5)

Step 4: Download and Parse Results

if status["status"] == "COMPLETED":
    download = requests.get(status["downloadUrl"])
    forecast_data = download.json()

    # Print summary
    for entry in forecast_data.get("entries", [])[:5]:
        print(f…

Clari Install & Auth

Overview

Set up Clari API access for exporting forecast data, pipeline snapshots, and revenue intelligence to your data warehouse. Clari uses API key authentication via the apikey header, with the primary API at api.clari.com/v4/.

Prerequisites

• Clari enterprise account with API access enabled
• Admin or RevOps role for API key generation
• Target data warehouse (Snowflake, BigQuery, or Redshift) for exports

Instructions

Step 1: Generate API Token

1. Log in to Clari at https://app.clari.com
2. Navigate to User Settings > API Token
3. Click Generate New API Token
4. Copy and store the token securely

# Store securely -- never commit
export CLARI_API_KEY="your-api-token-here"

# Verify the key works
curl -s -H "apikey: ${CLARI_API_KEY}" \
  https://api.clari.com/v4/export/forecast/list \
  | jq '.forecasts | length'

Step 2: Configure Environment

# .env -- NEVER commit this file
CLARI_API_KEY=your-api-token
CLARI_BASE_URL=https://api.clari.com/v4
CLARI_ORG_ID=your-org-id

# .gitignore
.env
.env.local

Step 3: Test API Connectivity

import requests
import os

api_key = os.environ["CLARI_API_KEY"]
headers = {"apikey": api_key, "Content-Type": "application/json"}

# List available forecasts
response = requests.get(
    "https://api.clari.com/v4/export/forecast/list",
    headers=headers,
)
response.raise_for_status()

forecasts = response.json()["forecasts"]
for fc in forecasts:
    print(f"  {fc['forecastName']} (ID: {fc['forecastId']})")

Step 4: Copilot API Setup (Optional)

Clari Copilot (conversation intelligence) has a separate API:

# Copilot uses OAuth2 -- different from the forecast API
# Register at https://api-doc.copilot.clari.com

export CLARI_COPILOT_CLIENT_ID="your-client-id"
export CLARI_COPILOT_CLIENT_SECRET="your-client-secret"

# Get access token
curl -X POST https://api.copilot.clari.com/oauth/token \
  -d "grant_type=client_credentials" \
  -d "client_id=${CLARI_COPILOT_CLIENT_ID}" \
  -d "client_secret=${CLARI_COPILOT_CLIENT_SECRET}"

Error Handling

Clari Local Dev Loop

Overview

Local development workflow for Clari integrations: mock forecast data for offline testing, schedule recurring exports, and build data transformation pipelines.

Prerequisites

• Completed clari-install-auth setup
• Python 3.10+ or Node.js 18+
• Local database or data warehouse access for testing

Instructions

Step 1: Project Structure

clari-integration/
├── src/
│   ├── clari_client.py       # API client wrapper
│   ├── export_pipeline.py    # Export and transform pipeline
│   ├── models.py             # Data models for forecast data
│   └── config.py             # Environment config
├── tests/
│   ├── fixtures/
│   │   ├── forecast_export.json    # Sample export response
│   │   └── job_status.json         # Sample job status
│   └── test_pipeline.py
├── .env.local                # Dev credentials (git-ignored)
├── .env.example
└── requirements.txt

Step 2: Mock Forecast Data for Testing

# tests/fixtures/forecast_export.json
MOCK_FORECAST = {
    "entries": [
        {
            "ownerName": "Jane Smith",
            "ownerEmail": "jane@example.com",
            "forecastAmount": 250000,
            "quotaAmount": 300000,
            "crmTotal": 180000,
            "crmClosed": 120000,
            "adjustmentAmount": 15000,
            "timePeriod": "2026_Q1"
        },
        {
            "ownerName": "Bob Johnson",
            "ownerEmail": "bob@example.com",
            "forecastAmount": 180000,
            "quotaAmount": 250000,
            "crmTotal": 140000,
            "crmClosed": 90000,
            "adjustmentAmount": 0,
            "timePeriod": "2026_Q1"
        }
    ]
}

Step 3: Test Pipeline Without API Calls

# tests/test_pipeline.py
import pytest
from src.export_pipeline import transform_forecast_data

def test_forecast_aggregation():
    data = MOCK_FORECAST
    result = transform_forecast_data(data)
    assert result["total_forecast"] == 430000
    assert result["total_quota"] == 550000
    assert result["attainment_percent"] == pytest.approx(78.2, rel=0.1)
    assert len(result["reps"]) == 2

def test_handles_empty_export():
    result = transform_forecast_data({"entries": []})
    assert result["total_forecast"] == 0

Step 4: Development Run Script

#!/bin/bash
# scripts/dev-export.sh
set -euo pipefail

source .env.local

echo "=== Clari Dev Export ==="
python3 src/export_pipeline.py \
  --forecast "company_forecast" \…

Clari Performance Tuning

Overview

Optimize Clari export pipelines: reduce export times, cache forecast data, and parallelize multi-period exports.

Prerequisites

• A scoped Clari API credential stored outside source control
• An approved forecast/export scope and named data owner
• A warehouse destination with retention and access controls
• A representative non-production or read-only validation window

Instructions

Parallel Multi-Period Export

from concurrent.futures import ThreadPoolExecutor, as_completed

def parallel_export(
    client,
    forecast_name: str,
    periods: list[str],
    max_workers: int = 3,
) -> dict[str, list[dict]]:
    results = {}

    def export_period(period: str) -> tuple[str, list[dict]]:
        data = client.export_and_download(forecast_name, period)
        return period, data.get("entries", [])

    with ThreadPoolExecutor(max_workers=max_workers) as executor:
        futures = {
            executor.submit(export_period, p): p for p in periods
        }
        for future in as_completed(futures):
            period, entries = future.result()
            results[period] = entries
            print(f"  {period}: {len(entries)} entries")

    return results

Cache Export Results

import json
import hashlib
from pathlib import Path
from datetime import datetime, timedelta

class ExportCache:
    def __init__(self, cache_dir: str = ".cache/clari", ttl_hours: int = 4):
        self.cache_dir = Path(cache_dir)
        self.cache_dir.mkdir(parents=True, exist_ok=True)
        self.ttl = timedelta(hours=ttl_hours)

    def _key(self, forecast: str, period: str) -> str:
        return hashlib.md5(f"{forecast}:{period}".encode()).hexdigest()

    def get(self, forecast: str, period: str) -> list[dict] | None:
        path = self.cache_dir / f"{self._key(forecast, period)}.json"
        if not path.exists():
            return None
        meta = json.loads(path.read_text())
        cached_at = datetime.fromisoformat(meta["cached_at"])
        if datetime.utcnow() - cached_at > self.ttl:
            return None
        return meta["entries"]

    def set(self, forecast: str, period: str, entries: list[dict]):
        path = self.cache_dir / f"{self._key(forecast, period)}.json"
        path.write_text(json.dumps({
            "cached_at": datetime.utcnow().isoformat(),
            "entries": entries,
        }))

Incremental Load to Warehouse

-- Use MERGE for incremental updates instead of full reload
MERGE INTO clari_forecasts AS target
USING staging_clari AS source
ON target.owner_email = source.owner_email
   AND target.time_period = source.time_period
   AND target.forecas…

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.

Prerequisites

• Approved production data owner, change record, and release window
• Separate production credential, state, and warehouse role
• Certified staging export and tested recovery from partial or stale data
• Monitoring, access-control, and retention policies accepted by owners

Instructions

Complete the checklist in sequence: verify identity and secret boundaries, validate API/export behavior with a controlled read, confirm idempotent loading and freshness alerting, then secure the required approvals before enabling the scheduled production job. Capture evidence for each item; do not promote a staging result or use a successful connection alone as production acceptance.

Authentication & Secrets

• [ ] CLARI_API_KEY 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

Clari Rate Limits

Overview

The Clari API enforces rate limits per API key. Export jobs are asynchronous and queued server-side, so the primary concern is polling frequency and concurrent export requests.

Prerequisites

• A scoped API token and approved forecast export scope
• Persistent job/attempt tracking for retry and idempotency decisions
• A scheduler that can defer work without spawning duplicate workers
• Monitoring for queue depth, response class, and terminal failures

Rate Limit Behavior

Instructions

Exponential Backoff for Export Polling

import time
import requests

def poll_with_backoff(
    job_id: str,
    api_key: str,
    max_attempts: int = 60,
    base_delay: float = 5.0,
    max_delay: float = 60.0,
) -> dict:
    for attempt in range(max_attempts):
        resp = requests.get(
            f"https://api.clari.com/v4/export/jobs/{job_id}",
            headers={"apikey": api_key},
        )

        if resp.status_code == 429:
            retry_after = int(resp.headers.get("Retry-After", base_delay))
            time.sleep(retry_after)
            continue

        resp.raise_for_status()
        status = resp.json()

        if status["status"] in ("COMPLETED", "FAILED"):
            return status

        delay = min(base_delay * (1.5 ** attempt), max_delay)
        time.sleep(delay)

    raise TimeoutError(f"Job {job_id} did not complete in {max_attempts} attempts")

Sequential Export Scheduler

def export_all_periods(
    client,
    forecast_name: str,
    periods: list[str],
    delay_between: float = 10.0,
) -> list[dict]:
    results = []
    for period in periods:
        print(f"Exporting {period}...")
        job = client.export_forecast(forecast_name, period)
        result = poll_with_backoff(job["jobId"], client.config.api_key)
        results.append(result)
        time.sleep(delay_between)  # Avoid hitting rate limits
    return results

Error Handling

Output

Return a per-job stat…

Clari Reference Architecture

Overview

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

Prerequisites

• Approved Clari export scope and a named revenue-data owner
• Separate development, staging, and production storage/state boundaries
• Warehouse access controls, retention policy, and audit logging
• An orchestrator capable of idempotent exports and reviewed recovery

Instructions

Build the pipeline in a non-production boundary first: pin the client and schema version, export a designated period, validate the manifest, and load through an idempotent warehouse operation. Promote the same reviewed design only after access controls, freshness alerts, and recovery behavior pass; keep individual forecast and owner data out of shared dashboards by default.

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_…

Clari SDK Patterns

Overview

Clari has no official SDK -- build typed wrappers around the v4 REST API. These patterns cover the Export API for forecasts, job polling, and data transformation pipelines.

Prerequisites

• Completed clari-install-auth setup
• Python 3.10+ (primary) or TypeScript 5+

Instructions

Step 1: Python Client

# clari_client.py
import os
import time
import requests
from dataclasses import dataclass, field
from typing import Optional

@dataclass
class ClariConfig:
    api_key: str
    base_url: str = "https://api.clari.com/v4"
    poll_interval: int = 5
    max_poll_attempts: int = 60

class ClariClient:
    def __init__(self, config: Optional[ClariConfig] = None):
        self.config = config or ClariConfig(
            api_key=os.environ["CLARI_API_KEY"]
        )
        self.session = requests.Session()
        self.session.headers.update({
            "apikey": self.config.api_key,
            "Content-Type": "text/plain",
        })

    def list_forecasts(self) -> list[dict]:
        resp = self.session.get(f"{self.config.base_url}/export/forecast/list")
        resp.raise_for_status()
        return resp.json()["forecasts"]

    def export_forecast(
        self,
        forecast_name: str,
        time_period: str,
        types: list[str] = None,
        currency: str = "USD",
        export_format: str = "JSON",
    ) -> dict:
        payload = {
            "timePeriod": time_period,
            "typesToExport": types or [
                "forecast", "quota", "forecast_updated",
                "adjustment", "crm_total", "crm_closed"
            ],
            "currency": currency,
            "schedule": "NONE",
            "includeHistorical": False,
            "exportFormat": export_format,
        }

        resp = self.session.post(
            f"{self.config.base_url}/export/forecast/{forecast_name}",
            json=payload,
        )
        resp.raise_for_status()
        return resp.json()

    def wait_for_job(self, job_id: str) -> dict:
        for attempt in range(self.config.max_poll_attempts):
            resp = self.session.get(
                f"{self.config.base_url}/export/jobs/{job_id}",
            )
            resp.raise_for_status()
            status = resp.json()

            if status["status"] == "COMPLETED":
                return status
            if status["status"] == "FAILED":
                raise ClariExportError(f"Job {job_id} failed: {status}")

            time.sleep(self.config.poll_interval)

        raise ClariExportError(f"Job {job_id} timed ou…

Clari Security Basics

Overview

Secure your Clari integration: API token management, exported data PII handling, and access control best practices.

Prerequisites

• An approved secret manager and named API-token owner
• A documented data classification for forecast and rep-level exports
• Role-based access groups for production and non-production consumers
• A tested token-rotation and incident escalation path

Instructions

Step 1: Token Management

# Store token in secrets manager
aws secretsmanager create-secret \
  --name "clari/prod/api-token" \
  --secret-string "${CLARI_API_KEY}"

# In CI/CD, load from secrets
export CLARI_API_KEY=$(aws secretsmanager get-secret-value \
  --secret-id "clari/prod/api-token" --query SecretString --output text)

Rotation: Clari API tokens are generated per-user. To rotate, generate a new token in User Settings, update all consumers, then discard the old one.

Step 2: Exported Data PII Handling

Clari export data contains PII (rep names, emails, deal amounts):

def redact_pii(entries: list[dict]) -> list[dict]:
    """Redact PII from forecast entries for non-production use."""
    import hashlib

    redacted = []
    for entry in entries:
        r = entry.copy()
        if "ownerEmail" in r:
            r["ownerEmail"] = hashlib.sha256(
                r["ownerEmail"].encode()
            ).hexdigest()[:12] + "@redacted"
        if "ownerName" in r:
            r["ownerName"] = f"Rep-{hashlib.sha256(r['ownerName'].encode()).hexdigest()[:6]}"
        redacted.append(r)
    return redacted

Step 3: Security Checklist

• [ ] API token in secrets manager, not in code
• [ ] .env files in .gitignore
• [ ] Exported data stored in access-controlled warehouse
• [ ] PII redacted in non-production environments
• [ ] Export download URLs are temporary -- do not cache
• [ ] Audit who has API token access
• [ ] Token regenerated if any team member leaves

Error Handling

Clari Upgrade & Migration

Overview

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

Prerequisites

• Current and target API/schema versions recorded in change control
• Representative non-production exports and an approved compatibility baseline
• Version-pinned client, warehouse migration, and rollback paths
• Data-owner approval for any field, retention, or analytics semantic change

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"))

Error Handling

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.

Prerequisites

• Authorized Clari/Copilot access and a registered HTTPS receiver
• Durable event/snapshot storage with bounded retention
• An allow-listed downstream notification target
• An approved threshold and escalation policy for forecast changes

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(…