Skip to main content

Documentation Index

Fetch the complete documentation index at: https://docs.anomalyarmor.ai/llms.txt

Use this file to discover all available pages before exploring further.

AnomalyArmor ships official SDKs for Python and TypeScript. Both talk to the same REST API (app.anomalyarmor.ai) with the same aa_live_* Bearer tokens, so an existing Python script and a new Vercel function see identical data.
LanguagePackageVersioningDocs
Pythonanomalyarmor-cli on PyPIpip install anomalyarmor-cliBelow
TypeScript / JavaScript@anomalyarmor/sdk on npmnpm install @anomalyarmor/sdkJavaScript SDK
The Python SDK also ships the anomalyarmor CLI. The TS SDK ships an npx anomalyarmor CLI. Both CLIs read ANOMALYARMOR_API_KEY from env as a convenience - library code in both SDKs requires the key to be passed explicitly.

​
Install side-by-side

pip install anomalyarmor-cli

​
Quickstart side-by-side

from anomalyarmor import Client

client = Client(api_key="aa_live_xxx")

# List assets
assets = client.assets.list(source="snowflake", limit=10)
for asset in assets:
    print(asset.qualified_name)

# Check freshness (raises StalenessError if stale)
client.freshness.require_fresh("snowflake.prod.warehouse.orders")

​
Python SDK

anomalyarmor-cli provides a Python SDK and CLI for programmatic access.

​
Installation

pip install anomalyarmor-cli
Requires Python 3.9 or higher.

​
Quick Start

from anomalyarmor import Client

# Initialize - uses ARMOR_API_KEY env var
client = Client()

# Or pass key directly
client = Client(api_key="aa_live_xxx")

# List assets
assets = client.assets.list(source="snowflake", limit=10)
for asset in assets:
    print(asset.qualified_name)

# Check freshness (raises StalenessError if stale)
client.freshness.require_fresh("snowflake.prod.warehouse.orders")

​
Configuration

​
Environment Variables

VariableDescription
ARMOR_API_KEYYour API key (recommended)
ARMOR_API_URLCustom API URL (optional)
export ARMOR_API_KEY="aa_live_your_key_here"

​
Client Options

from anomalyarmor import Client

client = Client(
    api_key="aa_live_xxx",      # Or use ARMOR_API_KEY env var
    api_url="https://...",       # Custom API URL (optional)
    timeout=30,                  # Request timeout in seconds
)

​
Resources

The client provides access to all AnomalyArmor resources: 
client.assets      # Data assets
client.freshness   # Freshness monitoring
client.schema      # Schema drift detection
client.lineage     # Data lineage
client.alerts      # Alert history
client.metrics     # Data quality metrics
client.api_keys    # API key management (admin scope)

​
Common Patterns

​
Airflow Pre-flight Check

Gate your pipeline on data freshness: 
from anomalyarmor import Client
from anomalyarmor.exceptions import StalenessError

def check_upstream_freshness():
    client = Client()

    try:
        # Raises StalenessError if stale
        client.freshness.require_fresh("snowflake.prod.warehouse.orders")
        print("Data is fresh, proceeding...")
    except StalenessError as e:
        print(f"Data is stale: {e}")
        raise  # Fail the task

​
List and Filter Assets

from anomalyarmor import Client

client = Client()

# All Snowflake tables
tables = client.assets.list(source="snowflake", type="table")

# Paginate through all assets
offset = 0
while True:
    assets = client.assets.list(limit=100, offset=offset)
    if not assets:
        break
    for asset in assets:
        process(asset)
    offset += 100

​
Trigger and Wait for Refresh

from anomalyarmor import Client

client = Client()

# Trigger freshness check and wait for completion
result = client.freshness.refresh(
    "snowflake.prod.warehouse.orders",
    wait=True  # Block until complete
)
print(f"Job {result.job_id}: {result.status}")

​
Check Lineage Before Running

from anomalyarmor import Client

client = Client()

# Get upstream dependencies
lineage = client.lineage.get("snowflake.prod.warehouse.orders")

# Check all upstream sources are fresh
for upstream in lineage.upstream:
    client.freshness.require_fresh(upstream.qualified_name)

print("All upstream sources are fresh!")

​
Monitor Data Quality Metrics

from anomalyarmor import Client

client = Client()
asset_id = "550e8400-e29b-41d4-a716-446655440000"

# Create a row count metric with anomaly detection
metric = client.metrics.create(
    asset_id,
    metric_type="row_count",
    table_path="snowflake.prod.warehouse.orders",
    capture_interval="daily",
    sensitivity=2.0,  # Alert on 2+ standard deviations
)

# Get metrics summary
summary = client.metrics.summary(asset_id)
print(f"Health: {summary.health_percentage}%")

# Check for recent anomalies
snapshots = client.metrics.snapshots(asset_id, metric.id, limit=7)
anomalies = [s for s in snapshots if s.is_anomaly]
if anomalies:
    print(f"Found {len(anomalies)} anomalies in the last 7 days")

​
Exception Handling

from anomalyarmor import Client
from anomalyarmor.exceptions import (
    StalenessError,      # Data is stale
    AuthenticationError, # Invalid/missing API key
    NotFoundError,       # Asset not found
    RateLimitError,      # Rate limit exceeded
    ValidationError,     # Invalid parameters
    ServerError,         # Server error
    ArmorError,          # Base exception
)

client = Client()

try:
    client.freshness.require_fresh("snowflake.prod.warehouse.orders")
except StalenessError as e:
    print(f"Data is stale: last updated {e.last_updated}")
except AuthenticationError:
    print("Invalid API key")
except RateLimitError as e:
    print(f"Rate limited, retry after {e.retry_after} seconds")
except ArmorError as e:
    print(f"API error: {e}")

​
Context Manager

The client supports context manager for automatic cleanup: 
from anomalyarmor import Client

with Client() as client:
    assets = client.assets.list()
    # Connection automatically closed

​
Type Hints

The SDK is fully typed for IDE support: 
from anomalyarmor import Client
from anomalyarmor.models import Asset, FreshnessStatus

client = Client()

asset: Asset = client.assets.get("snowflake.prod.warehouse.orders")
status: FreshnessStatus = client.freshness.get(asset.qualified_name)

print(status.is_fresh)  # IDE knows this is bool
print(status.last_updated)  # IDE knows this is datetime

​
Next Steps

SDK Reference

Complete method reference

CLI Guide

Command-line interface

Airflow Integration

Use in Airflow DAGs

API Reference

REST API documentation

​
Common Questions

​
Should I use the Python SDK or the TypeScript SDK?

Pick whichever matches the runtime you’re already using: Python for Airflow, dbt hooks, and notebooks; TypeScript for Next.js, Vercel functions, and Node services. Both SDKs wrap the same REST API and accept the same aa_live_* keys, so mixing them across services is fine. See the TypeScript SDK page for Node-specific setup.

​
Which Python version does the SDK require?

Python 3.9 or higher. The SDK ships fully typed models (Asset, FreshnessStatus, etc.) so you get IDE completion on every method and field. Install with pip install anomalyarmor-cli, which also installs the armor CLI.

​
How do I paginate through thousands of assets in Python?

Call client.assets.list(limit=100, offset=n) in a loop, incrementing offset by the page size until an empty page comes back. The SDK mirrors the REST pagination directly rather than hiding it, which keeps memory flat for very large accounts. The pattern is shown in the “List and Filter Assets” example above.

​
How do I use the SDK with Airflow?

Call client.freshness.require_fresh(asset) at the top of a task; it raises StalenessError when the asset is stale, which Airflow surfaces as a task failure. Set ARMOR_API_KEY as an Airflow connection secret and instantiate Client() with no arguments. The Airflow integration guide shows a full DAG.