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.

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

MethodEndpointDescription
GET/api/v1/sdk/alerts/summaryGet alerts summary
GET/api/v1/sdk/alertsList alerts
GET/api/v1/sdk/alerts/rulesList 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"

​
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

ParameterTypeDefaultDescription
statusstring-Filter: triggered, acknowledged, resolved
alert_typestring-Filter: freshness, schema_change, row_count
asset_idstring-Filter by asset qualified name
sincedatetime-Alerts since timestamp
limitinteger50Max results
offsetinteger0Results to skip
curl -H "Authorization: Bearer aa_live_xxx" \
  "https://api.anomalyarmor.ai/api/v1/sdk/alerts?status=triggered"

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

​
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

TypeDescription
freshnessData is stale beyond threshold
schema_changeSchema change detected
row_countRow count anomaly detected
null_percentageNull percentage exceeded threshold
duplicate_keysDuplicate primary keys detected

​
Alert Statuses

StatusDescription
triggeredAlert is active and needs attention
acknowledgedAlert has been seen but not resolved
resolvedAlert 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.