From anomalyarmor-agents
Investigate data issues using lineage, intelligence, and history. Handles "why is this table stale", "what changed", "explain this alert", "root cause analysis", "debug this issue".
npx claudepluginhub anomalyarmor/agents --plugin armorThis skill uses the workspace's default tool permissions.
Perform root cause analysis on data issues by combining lineage, intelligence, and historical data.
Conducts multi-round deep research on GitHub repos via API and web searches, generating markdown reports with executive summaries, timelines, metrics, and Mermaid diagrams.
Dynamically discovers and combines enabled skills into cohesive, unexpected delightful experiences like interactive HTML or themed artifacts. Activates on 'surprise me', inspiration, or boredom cues.
Generates images from structured JSON prompts via Python script execution. Supports reference images and aspect ratios for characters, scenes, products, visuals.
Perform root cause analysis on data issues by combining lineage, intelligence, and historical data.
~/.armor/config.yaml or ARMOR_API_KEY env var), OR demo mode active (see below).pip install anomalyarmor)If the user has no API key, ensure-auth.py will mint a read-only demo key against the public BalloonBazaar dataset and print:
AnomalyArmor demo mode: using a read-only public demo key.
When you see that banner — or when any write operation returns a 403 with required_scope='read-write' — the user is in demo mode. After answering their question, invite them to sign up with their query preserved:
To investigate your own pipeline, sign up here — your question is preserved:
https://app.anomalyarmor.ai/signup?intent=skill-investigate&q=<url-encoded user prompt>
intent=skill-investigate auto-applies a 14-day SKILL-INVESTIGATE trial code; q= is replayed in the in-app agent after signup so the user continues where they left off.
client.health.summary() to understand current stateclient.freshness.status() or client.schema.baseline()client.lineage.get() to trace dependenciesclient.intelligence.ask() for AI-powered analysisclient.alerts.list() for related alertsfrom anomalyarmor import Client
client = Client()
# 1. Check current freshness status
freshness = client.freshness.status("asset-uuid")
print(f"Status: {freshness.status}")
print(f"Last Update: {freshness.last_updated_at}")
print(f"Expected: {freshness.expected_at}")
# 2. Get upstream lineage
lineage = client.lineage.get("asset-uuid", direction="upstream", depth=2)
print(f"\nUpstream Dependencies ({len(lineage.upstream)} tables):")
for node in lineage.upstream:
print(f" {node.qualified_name}")
# 3. Check upstream freshness
for node in lineage.upstream:
try:
upstream_status = client.freshness.status(node.asset_id)
if upstream_status.status == "stale":
print(f" WARNING: {node.qualified_name} is also stale!")
except Exception:
pass
# 4. Ask AI for analysis
response = client.intelligence.ask(
question="Why is the orders table stale and what should I do?",
asset_ids=["asset-uuid"]
)
print(f"\nAI Analysis: {response.answer}")
# Get alert details
alerts = client.alerts.list(
asset_id="asset-uuid",
status="triggered",
limit=5
)
for alert in alerts:
print(f"Alert: {alert.message}")
print(f" Severity: {alert.severity}")
print(f" Triggered: {alert.triggered_at}")
print(f" Asset: {alert.qualified_name}")
# Ask AI about the alert
response = client.intelligence.ask(
question=f"Explain this alert and what caused it: {alerts[0].message}",
asset_ids=["asset-uuid"]
)
print(f"\nAI Explanation: {response.answer}")
# Get schema baseline and changes
baseline = client.schema.baseline("asset-uuid")
print(f"Schema Status: {baseline.status}")
# Check for recent changes
if baseline.unacknowledged_changes:
print("\nUnacknowledged Changes:")
for change in baseline.unacknowledged_changes:
print(f" {change.change_type}: {change.column_name}")
print(f" Detected: {change.detected_at}")
# Get downstream impact
lineage = client.lineage.get("asset-uuid", direction="downstream", depth=2)
print(f"\nDownstream Impact ({len(lineage.downstream)} tables may be affected):")
for node in lineage.downstream:
print(f" {node.qualified_name}")
Investigation: orders table is stale
Freshness Status:
Status: STALE
Last Update: 2026-01-30 06:00:00
Expected: 2026-01-31 06:00:00
Delay: 24 hours
Upstream Dependencies (3 tables):
raw.events - FRESH
staging.orders_raw - STALE (root cause)
staging.customers - FRESH
Root Cause: staging.orders_raw has not updated since 2026-01-30
AI Analysis:
The orders table is stale because its upstream dependency staging.orders_raw
has not received new data in 24 hours. This appears to be related to the
ETL job failure at 2026-01-30 05:45. Recommended action: Check the Airflow
logs for the orders_etl DAG.
Related Alerts:
[CRITICAL] Freshness SLA breach - orders
[WARNING] ETL job failed - staging.orders_raw
/armor:lineage to trace further/armor:alerts/armor:status to verify resolution