Claude AI Agents for Data Analysis: Automated Reporting

from anthropic import Anthropic
from anthropic.lib.agents import Agent, tool
import snowflake.connector
import pandas as pd
from typing import Any

client = Anthropic()

# Initialize Snowflake connection pool
snowflake_conn = snowflake.connector.connect(
    user="service_account",
    password="...",
    account="xy12345.us-east-1",
    warehouse="analytics_wh",
    database="production"
)

@tool
def query_revenue_data(
    metric: str,
    time_period: str,
    groupby: str = "day"
) -> dict:
    """
    Query revenue metrics from the data warehouse.

    Args:
        metric: One of 'total_revenue', 'arpu', 'churn_rate'
        time_period: e.g. 'last_7_days', 'last_30_days', 'ytd'
        groupby: Aggregation level ('day', 'week', 'month')
    Returns:
        Dict with timeseries data and summary statistics
    """
    # Build safe parameterized query
    period_map = {
        'last_7_days': "CURRENT_DATE - 7",
        'last_30_days': "CURRENT_DATE - 30",
        'ytd': "DATE_TRUNC('YEAR', CURRENT_DATE)"
    }

    query = f"""
    SELECT
        DATE_TRUNC('{groupby}', order_date) as period,
        {metric} as value,
        COUNT(*) as record_count
    FROM analytics.transactions
    WHERE order_date >= {period_map[time_period]}
    GROUP BY period
    ORDER BY period DESC
    LIMIT 100
    """

    # Execute with timeout protection
    cursor = snowflake_conn.cursor()
    cursor.execute(f"SELECT COUNT(*) FROM analytics.transactions WHERE order_date >= {period_map[time_period]}")
    row_count = cursor.fetchone()[0]

    if row_count > 10_000_000:
        cursor.execute(query.replace("analytics.transactions",
                                    f"(SELECT * FROM analytics.transactions TABLESAMPLE (1 percent)) as sampled"))
    else:
        cursor.execute(query)

    results = cursor.fetchall()
    df = pd.DataFrame(results, columns=['period', 'value', 'count'])

    return {
        "data": df.to_dict('records'),
        "summary": {
            "min": float(df['value'].min()),
            "max": float(df['value'].max()),
            "mean": float(df['value'].mean()),
            "trend": "up" if df['value'].iloc[-1] > df['value'].iloc[0] else "down"
        }
    }

@tool
def detect_anomalies(
    metric: str,
    lookback_days: int = 30
) -> dict:
    """
    Detect statistical anomalies in a metric using z-score analysis.
    """
    cursor = snowflake_conn.cursor()
    cursor.execute(f"""
    WITH daily_metrics AS (
        SELECT
            DATE(order_date) as day,
            {metric} as value
        FROM analytics.transactions
        WHERE order_date >= CURRENT_DATE - {lookback_days}
    ),
    stats AS (
        SELECT
            AVG(value) as mean_val,
            STDDEV(value) as std_val
        FROM daily_metrics
    )
    SELECT
        day,
        value,
        ABS((value - stats.mean_val) / NULLIF(stats.std_val, 0)) as z_score
    FROM daily_metrics, stats
    WHERE ABS((value - stats.mean_val) / NULLIF(stats.std_val, 0)) > 2
    ORDER BY day DESC
    """)

    anomalies = cursor.fetchall()
    return {
        "count": len(anomalies),
        "anomalies": [{"date": a[0], "value": a[1], "z_score": a[2]} for a in anomalies]
    }

# Define the reporting agent
data_analyst_agent = Agent(
    client=client,
    model="claude-sonnet-4-6",
    system="""You are a data analyst agent. Your role is to:
1. Help stakeholders understand business metrics through data analysis
2. Execute queries safely using available tools
3. Detect anomalies and trends in the data
4. Provide clear, actionable insights in business language
5. Always validate data before drawing conclusions

When generating reports:
- Start with executive summary (1-2 sentences)
- Include specific metrics and comparisons
- Flag anomalies or concerning trends
- End with one recommended action
- Never make up data — only report what you query""",
    tools=[query_revenue_data, detect_anomalies],
    max_tokens=2048,
    max_iterations=10
)

# Run an analysis
result = data_analyst_agent.run(
    "Generate a revenue report for the last 30 days by week. Flag any unusual trends."
)
print(result.output)

def validate_query_result(result: dict, schema: dict, name: str) -> bool:
    """Validate result before using in report"""

    # Check required columns
    for col in schema['required_columns']:
        if col not in result.get('columns', []):
            raise ValueError(f"Missing column {col} in {name}")

    # Check row count
    row_count = len(result.get('data', []))
    if row_count == 0:
        raise ValueError(f"Query returned 0 rows for {name}")
    if row_count > schema['max_rows']:
        raise ValueError(f"Query exceeded max_rows ({row_count} > {schema['max_rows']})")

    # Check value ranges
    for col in schema.get('numeric_columns', []):
        values = [row.get(col) for row in result['data'] if col in row]
        if any(v for v in values if v is not None and v < schema['min_value']):
            raise ValueError(f"Column {col} contains value below minimum")

    return True