Connect

Skill anomalyarmor/agents/skills/connect

Connect a new data source to AnomalyArmor. Handles "connect my database", "add snowflake", "setup postgres", "connect warehouse".From its SKILL.md

Install

npx -y skills add anomalyarmor/agents --skill connect

Assembled from the repository path, not quoted from the project. Check it against their README if it does not work.

One thing to look at

1 stars1 stars. Stars are a popularity signal and not a quality one, but at this level it is likely that nobody has read this closely except its author, and you would be relying on your own review.

SKILL.md

4.3 KB, as published. Nobody here has run it

Connect Data Source

Guide users through connecting a new data source (database, warehouse) to AnomalyArmor.

Prerequisites

AnomalyArmor API key configured (~/.armor/config.yaml or ARMOR_API_KEY env var)
Python SDK installed (pip install anomalyarmor)
Database credentials ready

When to Use

"Connect my Snowflake warehouse"
"Add my PostgreSQL database"
"Setup Databricks connection"
"Connect a new data source"

Supported Data Sources

snowflake - Snowflake Data Cloud
postgresql - PostgreSQL
databricks - Databricks
bigquery - Google BigQuery
redshift - Amazon Redshift
mysql - MySQL
clickhouse - ClickHouse

Steps

Ask user for data source type if not specified
Collect connection configuration (credentials, host, database, etc.)
Create the asset with client.assets.create()
Test the connection with client.assets.test_connection()
If successful, trigger schema discovery with client.assets.trigger_discovery()
Track discovery progress with client.jobs.status()

Example Usage

Connect Snowflake

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')}")

Connect PostgreSQL

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

Connect Databricks

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..."
    }
)

Connection Config by Source Type

Snowflake

{
    "account": "abc123.us-east-1",
    "warehouse": "COMPUTE_WH",
    "database": "ANALYTICS",
    "user": "user",
    "password": "password",
    "role": "ROLE_NAME"  # optional
}

PostgreSQL

{
    "host": "hostname",
    "port": 5432,
    "database": "dbname",
    "user": "user",
    "password": "password",
    "sslmode": "require"  # optional
}

BigQuery

{
    "project_id": "my-project",
    "credentials_json": "{...}"  # Service account JSON
}

Security Notes

Never hardcode credentials in scripts
Use environment variables or secure vaults
Create read-only database users for AnomalyArmor
Ensure network connectivity (whitelist IPs if needed)

Follow-up Actions

After connecting, use:

/armor:monitor to set up freshness and schema monitoring
/armor:analyze to generate AI intelligence
/armor:status to verify health