From anomalyarmor-agents
Connect a new data source to AnomalyArmor. Handles "connect my database", "add snowflake", "setup postgres", "connect warehouse".
npx claudepluginhub anomalyarmor/agents --plugin armorThis skill uses the workspace's default tool permissions.
Guide users through connecting a new data source (database, warehouse) to AnomalyArmor.
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.
Guide users through connecting a new data source (database, warehouse) to AnomalyArmor.
~/.armor/config.yaml or ARMOR_API_KEY env var)pip install anomalyarmor)snowflake - Snowflake Data Cloudpostgresql - PostgreSQLdatabricks - Databricksbigquery - Google BigQueryredshift - Amazon Redshiftmysql - MySQLclickhouse - ClickHouseclient.assets.create()client.assets.test_connection()client.assets.trigger_discovery()client.jobs.status()from anomalyarmor import Client
import time
client = Client()
# Create the asset
asset = client.assets.create(
name="Analytics Warehouse",
source_type="snowflake",
connection_config={
"account": "abc123.us-east-1",
"warehouse": "COMPUTE_WH",
"database": "ANALYTICS",
"user": "anomalyarmor_user",
"password": "your_password", # Or use key pair auth
"role": "ANOMALYARMOR_ROLE" # Optional
},
description="Main analytics data warehouse"
)
print(f"Created asset: {asset.id}")
# Test connection
result = client.assets.test_connection(asset.id)
if result.success:
print("Connection successful!")
else:
print(f"Connection failed: {result.error_message}")
exit(1)
# Trigger schema discovery
job = client.assets.trigger_discovery(asset.id)
print(f"Discovery started: {job.job_id}")
# Wait for discovery to complete
while True:
status = client.jobs.status(job.job_id)
print(f"Discovery progress: {status.get('progress', 0)}%")
if status.get('status') in ('completed', 'failed'):
break
time.sleep(5)
if status.get('status') == 'completed':
print("Schema discovery complete!")
else:
print(f"Discovery failed: {status.get('error')}")
from anomalyarmor import Client
client = Client()
asset = client.assets.create(
name="Production Database",
source_type="postgresql",
connection_config={
"host": "db.example.com",
"port": 5432,
"database": "production",
"user": "readonly_user",
"password": "your_password",
"sslmode": "require"
}
)
# Test and discover...
from anomalyarmor import Client
client = Client()
asset = client.assets.create(
name="Databricks Lakehouse",
source_type="databricks",
connection_config={
"host": "adb-1234567890.1.azuredatabricks.net",
"http_path": "/sql/1.0/warehouses/abc123",
"access_token": "dapi..."
}
)
{
"account": "abc123.us-east-1",
"warehouse": "COMPUTE_WH",
"database": "ANALYTICS",
"user": "user",
"password": "password",
"role": "ROLE_NAME" # optional
}
{
"host": "hostname",
"port": 5432,
"database": "dbname",
"user": "user",
"password": "password",
"sslmode": "require" # optional
}
{
"project_id": "my-project",
"credentials_json": "{...}" # Service account JSON
}
After connecting, use:
/armor:monitor to set up freshness and schema monitoring/armor:analyze to generate AI intelligence/armor:status to verify health