Alerts API

The Alerts API provides access to data quality alerts triggered by AnomalyArmor monitoring. Use it to query alert history, check active incidents, and integrate with your incident response workflows.

Endpoints

Method	Endpoint	Description
GET	`/api/v1/sdk/alerts/summary`	Get alerts summary
GET	`/api/v1/sdk/alerts`	List alerts
GET	`/api/v1/sdk/alerts/rules`	List alert rules

Get Alerts Summary

GET /api/v1/sdk/alerts/summary

Returns aggregate alert statistics.

curl -H "Authorization: Bearer aa_live_xxx" \
  "https://api.anomalyarmor.ai/api/v1/sdk/alerts/summary"

from anomalyarmor import Client

client = Client()
summary = client.alerts.summary()
print(f"Triggered: {summary.triggered}")
print(f"Last 24h: {summary.triggered_last_24h}")

armor alerts summary

Response

{
  "data": {
    "total": 45,
    "by_status": {
      "triggered": 5,
      "acknowledged": 3,
      "resolved": 37
    },
    "by_type": {
      "freshness": 20,
      "schema_change": 15,
      "row_count": 10
    },
    "triggered_last_24h": 8,
    "triggered_last_7d": 25
  }
}

List Alerts

GET /api/v1/sdk/alerts

Query Parameters

Parameter	Type	Default	Description
`status`	string	-	Filter: `triggered`, `acknowledged`, `resolved`
`alert_type`	string	-	Filter: `freshness`, `schema_change`, `row_count`
`asset_id`	string	-	Filter by asset qualified name
`since`	datetime	-	Alerts since timestamp
`limit`	integer	50	Max results
`offset`	integer	0	Results to skip

curl -H "Authorization: Bearer aa_live_xxx" \
  "https://api.anomalyarmor.ai/api/v1/sdk/alerts?status=triggered"

from anomalyarmor import Client

client = Client()

# List all alerts
alerts = client.alerts.list()

# Filter by status
triggered = client.alerts.list(status="triggered")
for alert in triggered:
    print(f"ACTIVE: {alert.qualified_name} - {alert.message}")

# Filter by asset
asset_alerts = client.alerts.list(
    asset_id="snowflake.prod.warehouse.orders"
)

# Combined filters
critical = client.alerts.list(
    status="triggered",
    alert_type="freshness"
)

armor alerts list
armor alerts list --status triggered
armor alerts list --asset snowflake.prod.warehouse.orders
armor alerts list --type freshness

Response

{
  "data": [
    {
      "id": "alert_123",
      "alert_type": "freshness",
      "status": "triggered",
      "asset_id": "550e8400-e29b-41d4-a716-446655440000",
      "qualified_name": "snowflake.prod.warehouse.orders",
      "message": "Data is 26 hours stale (threshold: 24 hours)",
      "triggered_at": "2024-12-04T08:00:00Z",
      "acknowledged_at": null,
      "resolved_at": null,
      "details": {
        "last_updated": "2024-12-03T06:00:00Z",
        "threshold_hours": 24,
        "hours_stale": 26
      },
      "rule_id": "rule_freshness_orders"
    }
  ],
  "pagination": {
    "total": 5,
    "limit": 50,
    "offset": 0,
    "has_more": false
  }
}

List Alert Rules

GET /api/v1/sdk/alerts/rules

Returns configured alert rules for your organization.

curl -H "Authorization: Bearer aa_live_xxx" \
  "https://api.anomalyarmor.ai/api/v1/sdk/alerts/rules"

from anomalyarmor import Client

client = Client()
rules = client.alerts.rules()
for rule in rules:
    status = "enabled" if rule.enabled else "disabled"
    print(f"{rule.name}: {rule.alert_type} ({status})")

Response

{
  "data": [
    {
      "id": "rule_freshness_orders",
      "name": "Orders Freshness",
      "alert_type": "freshness",
      "asset_id": "snowflake.prod.warehouse.orders",
      "enabled": true,
      "config": {
        "threshold_hours": 24
      },
      "destinations": ["slack", "email"],
      "created_at": "2024-01-15T10:00:00Z"
    }
  ],
  "pagination": {
    "total": 15,
    "limit": 50,
    "offset": 0,
    "has_more": false
  }
}

Alert Types

Type	Description
`freshness`	Data is stale beyond threshold
`schema_change`	Schema change detected
`row_count`	Row count anomaly detected
`null_percentage`	Null percentage exceeded threshold
`duplicate_keys`	Duplicate primary keys detected

Alert Statuses

Status	Description
`triggered`	Alert is active and needs attention
`acknowledged`	Alert has been seen but not resolved
`resolved`	Alert has been resolved

Use Case: Incident Response

Build an incident response script:

from anomalyarmor import Client

client = Client()

def get_active_incidents():
    """Get all active alerts for incident response."""
    alerts = client.alerts.list(status="triggered")

    by_type = {"schema": [], "freshness": [], "discovery": []}
    for alert in alerts:
        by_type[alert.alert_type].append(alert)

    print("=== Active Incidents ===")
    for alert_type in ["schema", "freshness", "discovery"]:
        if by_type[alert_type]:
            print(f"\n{alert_type.upper()} ({len(by_type[alert_type])}):")
            for alert in by_type[alert_type]:
                print(f"  - {alert.qualified_name}: {alert.message}")

    return alerts

incidents = get_active_incidents()

Output:

=== Active Incidents ===

SCHEMA (1):
  - snowflake.prod.warehouse.payments: Column removed: deprecated_field

FRESHNESS (2):
  - snowflake.prod.warehouse.orders: Data is 26 hours stale
  - bigquery.analytics.events: Data is 4 hours stale

Use Case: Slack Bot Integration

Post alerts to Slack:

from anomalyarmor import Client
import requests

client = Client()
SLACK_WEBHOOK = "https://hooks.slack.com/services/..."

def post_alerts_to_slack():
    """Post triggered alerts to Slack."""
    alerts = client.alerts.list(status="triggered")

    if not alerts:
        return

    blocks = [{"type": "header", "text": {"type": "plain_text", "text": f"{len(alerts)} Active Alerts"}}]

    for alert in alerts[:5]:  # Limit to 5
        blocks.append({
            "type": "section",
            "text": {
                "type": "mrkdwn",
                "text": f"*{alert.qualified_name}*\n{alert.message}"
            }
        })

    requests.post(SLACK_WEBHOOK, json={"blocks": blocks})

# Run periodically
post_alerts_to_slack()

Common Questions

How do I filter alerts by asset or type?

Combine asset_id, alert_type, and status query parameters on GET /api/v1/sdk/alerts. For example, ?status=triggered&alert_type=freshness returns only active freshness alerts. The since parameter accepts an ISO-8601 timestamp to bound history queries.

Can I acknowledge or resolve alerts through the API?

The read endpoints on this page cover history and summaries. For state transitions (acknowledge / resolve), use the per-alert endpoints documented in the TypeScript SDK reference (client.alerts.acknowledge, client.alerts.resolve) which post to /api/v1/alerts/{alert_id}/acknowledge and .../resolve.

What alert types does AnomalyArmor emit?

The API returns freshness, schema_change, row_count, null_percentage, and duplicate_keys. Each carries a details object with the relevant context (e.g. hours_stale, threshold_hours for freshness). Use alert_type to route to different Slack channels or PagerDuty services.

API Overview

API Reference

API Guides

SDKs

CLI

Integrations

AI Agents

Endpoints

Get Alerts Summary

Response

List Alerts

Query Parameters

Response

List Alert Rules

Response

Alert Types

Alert Statuses

Use Case: Incident Response

Use Case: Slack Bot Integration

Common Questions

How do I filter alerts by asset or type?

Can I acknowledge or resolve alerts through the API?

What alert types does AnomalyArmor emit?

​Endpoints

​Get Alerts Summary

​Response

​List Alerts

​Query Parameters

​Response

​List Alert Rules

​Response

​Alert Types

​Alert Statuses

​Use Case: Incident Response

​Use Case: Slack Bot Integration

​Common Questions

​How do I filter alerts by asset or type?

​Can I acknowledge or resolve alerts through the API?

​What alert types does AnomalyArmor emit?

Endpoints

Get Alerts Summary

Response

List Alerts

Query Parameters

Response

List Alert Rules

Response

Alert Types

Alert Statuses

Use Case: Incident Response

Use Case: Slack Bot Integration

Common Questions

How do I filter alerts by asset or type?

Can I acknowledge or resolve alerts through the API?

What alert types does AnomalyArmor emit?