> ## 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.

# SDK overview

> Get started with the AnomalyArmor SDKs (Python and TypeScript)

<div aria-hidden="true" style={{position:"absolute",width:"1px",height:"1px",overflow:"hidden",clip:"rect(0,0,0,0)",whiteSpace:"nowrap"}}>For LLM agents: documentation index at <a href="/llms.txt" tabIndex={-1}>/llms.txt</a>, full text at <a href="/llms-full.txt" tabIndex={-1}>/llms-full.txt</a>. Append .md to any page URL for plain markdown.</div>
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.

| Language                | Package                    | Versioning                      | Docs                              |
| ----------------------- | -------------------------- | ------------------------------- | --------------------------------- |
| Python                  | `anomalyarmor-cli` on PyPI | `pip install anomalyarmor-cli`  | Below                             |
| TypeScript / JavaScript | `@anomalyarmor/sdk` on npm | `npm install @anomalyarmor/sdk` | [JavaScript SDK](/sdk/javascript) |

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

<CodeGroup>
  ```bash Python theme={null}
  pip install anomalyarmor-cli
  ```

  ```bash TypeScript theme={null}
  npm install @anomalyarmor/sdk
  ```
</CodeGroup>

## Quickstart side-by-side

<CodeGroup>
  ```python Python theme={null}
  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")
  ```

  ```ts TypeScript theme={null}
  import { createAnomalyArmorClient } from '@anomalyarmor/sdk';

  const client = createAnomalyArmorClient({ apiKey: 'aa_live_xxx' });

  // Pull an alert overview
  const overview = await client.alerts.overview();
  console.log(overview);

  // Check freshness for an asset (throws AnomalyArmorApiError on 4xx/5xx)
  const freshness = await client.freshness.check('asset-uuid');
  ```
</CodeGroup>

## Python SDK

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

### Installation

```bash theme={null}
pip install anomalyarmor-cli
```

<Note>
  Requires Python 3.9 or higher.
</Note>

### Quick Start

```python theme={null}
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

| Variable        | Description                |
| --------------- | -------------------------- |
| `ARMOR_API_KEY` | Your API key (recommended) |
| `ARMOR_API_URL` | Custom API URL (optional)  |

```bash theme={null}
export ARMOR_API_KEY="aa_live_your_key_here"
```

### Client Options

```python theme={null}
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:

```python theme={null}
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.investigations  # Root-cause investigations (EvidenceCapsule)
client.api_keys        # API key management (admin scope)
```

## Common Patterns

### Airflow Pre-flight Check

Gate your pipeline on data freshness:

```python theme={null}
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

```python theme={null}
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

```python theme={null}
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

```python theme={null}
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

```python theme={null}
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

```python theme={null}
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:

```python theme={null}
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:

```python theme={null}
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

<CardGroup cols={2}>
  <Card title="SDK Reference" icon="book" href="/sdk/reference">
    Complete method reference
  </Card>

  <Card title="CLI Guide" icon="terminal" href="/cli/overview">
    Command-line interface
  </Card>

  <Card title="Airflow Integration" icon="wind" href="/integrations/airflow">
    Use in Airflow DAGs
  </Card>

  <Card title="API Reference" icon="code" href="/api/overview">
    REST API documentation
  </Card>
</CardGroup>

## 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](/sdk/javascript) 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](/integrations/airflow) shows a full DAG.