How to Integrate AI Agents with Snowflake

pip install snowflake-connector-python langchain langchain-openai sqlalchemy snowflake-sqlalchemy python-dotenv

export SNOWFLAKE_ACCOUNT="your-account-identifier"  # e.g., xy12345.us-east-1
export SNOWFLAKE_USER="your-username"
export SNOWFLAKE_PASSWORD="your-password"
export SNOWFLAKE_WAREHOUSE="COMPUTE_WH"
export SNOWFLAKE_DATABASE="ANALYTICS"
export SNOWFLAKE_SCHEMA="PUBLIC"
export SNOWFLAKE_ROLE="ANALYST_ROLE"  # Use a read-only role

import snowflake.connector
import os

conn = snowflake.connector.connect(
    account=os.getenv("SNOWFLAKE_ACCOUNT"),
    user=os.getenv("SNOWFLAKE_USER"),
    password=os.getenv("SNOWFLAKE_PASSWORD"),
    warehouse=os.getenv("SNOWFLAKE_WAREHOUSE"),
    database=os.getenv("SNOWFLAKE_DATABASE"),
    schema=os.getenv("SNOWFLAKE_SCHEMA"),
    role=os.getenv("SNOWFLAKE_ROLE")
)
cursor = conn.cursor()
cursor.execute("SELECT CURRENT_VERSION()")
print(cursor.fetchone())  # Should print Snowflake version

import os
import pandas as pd
import snowflake.connector
from langchain.tools import tool
from dotenv import load_dotenv

load_dotenv()


def get_snowflake_connection():
    """Get a Snowflake database connection."""
    return snowflake.connector.connect(
        account=os.getenv("SNOWFLAKE_ACCOUNT"),
        user=os.getenv("SNOWFLAKE_USER"),
        password=os.getenv("SNOWFLAKE_PASSWORD"),
        warehouse=os.getenv("SNOWFLAKE_WAREHOUSE"),
        database=os.getenv("SNOWFLAKE_DATABASE"),
        schema=os.getenv("SNOWFLAKE_SCHEMA"),
        role=os.getenv("SNOWFLAKE_ROLE")
    )


def execute_query(sql: str, max_rows: int = 100) -> pd.DataFrame:
    """Execute a SQL query and return results as a DataFrame."""
    conn = get_snowflake_connection()
    try:
        cursor = conn.cursor()
        cursor.execute(sql)
        columns = [col[0] for col in cursor.description]
        rows = cursor.fetchmany(max_rows)
        return pd.DataFrame(rows, columns=columns)
    finally:
        conn.close()


@tool
def query_snowflake(sql: str) -> str:
    """
    Execute a SELECT SQL query against Snowflake and return results.
    Only SELECT queries are permitted. Always LIMIT results to avoid excessive data transfer.
    """
    # Safety: only allow SELECT queries
    sql_clean = sql.strip().upper()
    if not sql_clean.startswith("SELECT") and not sql_clean.startswith("WITH"):
        return "Error: Only SELECT queries are permitted for safety reasons"

    # Add LIMIT if not present
    if "LIMIT" not in sql_clean:
        sql = sql.rstrip(";") + " LIMIT 100"

    df = execute_query(sql)
    if df.empty:
        return "Query returned no results"

    return f"Results ({len(df)} rows):\n{df.to_string(index=False, max_rows=20)}"


@tool
def get_table_schema(table_name: str) -> str:
    """Get the schema (column names and types) for a Snowflake table."""
    # Parse database.schema.table format
    parts = table_name.split(".")
    if len(parts) == 3:
        db, schema, table = parts
    elif len(parts) == 2:
        db = os.getenv("SNOWFLAKE_DATABASE")
        schema, table = parts
    else:
        db = os.getenv("SNOWFLAKE_DATABASE")
        schema = os.getenv("SNOWFLAKE_SCHEMA")
        table = table_name

    sql = f"""
    SELECT COLUMN_NAME, DATA_TYPE, IS_NULLABLE, CHARACTER_MAXIMUM_LENGTH
    FROM {db}.INFORMATION_SCHEMA.COLUMNS
    WHERE TABLE_SCHEMA = '{schema.upper()}' AND TABLE_NAME = '{table.upper()}'
    ORDER BY ORDINAL_POSITION"""

    df = execute_query(sql)
    if df.empty:
        return f"Table {table_name} not found or no columns returned"
    return f"Schema for {table_name}:\n{df.to_string(index=False)}"


@tool
def list_tables(schema: str = None) -> str:
    """List all tables in the current database schema."""
    target_schema = schema or os.getenv("SNOWFLAKE_SCHEMA")
    sql = f"""
    SELECT TABLE_NAME, TABLE_TYPE, ROW_COUNT, BYTES
    FROM {os.getenv('SNOWFLAKE_DATABASE')}.INFORMATION_SCHEMA.TABLES
    WHERE TABLE_SCHEMA = '{target_schema.upper()}'
    ORDER BY TABLE_NAME"""

    df = execute_query(sql)
    if df.empty:
        return f"No tables found in schema {target_schema}"
    return f"Tables in {target_schema}:\n{df.to_string(index=False)}"


@tool
def detect_metric_anomaly(table: str, metric_column: str, date_column: str,
                          lookback_days: int = 30) -> str:
    """
    Detect anomalies in a time-series metric by comparing recent values
    to the historical mean and standard deviation.
    """
    sql = f"""
    WITH historical AS (
        SELECT
            AVG({metric_column}) AS avg_val,
            STDDEV({metric_column}) AS std_val
        FROM {table}
        WHERE {date_column} >= DATEADD(day, -{lookback_days}, CURRENT_DATE)
          AND {date_column} < DATEADD(day, -3, CURRENT_DATE)
    ),
    recent AS (
        SELECT
            {date_column},
            {metric_column},
            AVG({metric_column}) OVER () AS recent_avg
        FROM {table}
        WHERE {date_column} >= DATEADD(day, -3, CURRENT_DATE)
    )
    SELECT
        r.{date_column},
        r.{metric_column},
        h.avg_val AS historical_avg,
        h.std_val AS historical_std,
        (r.{metric_column} - h.avg_val) / NULLIF(h.std_val, 0) AS z_score
    FROM recent r, historical h
    ORDER BY r.{date_column} DESC"""

    df = execute_query(sql)
    if df.empty:
        return "No data available for anomaly detection"

    anomalies = df[df["Z_SCORE"].abs() > 2] if "Z_SCORE" in df.columns else pd.DataFrame()
    if anomalies.empty:
        return f"No significant anomalies detected in {metric_column} over the last {lookback_days} days"

    return f"Anomalies detected in {metric_column}:\n{anomalies.to_string(index=False)}"

from langchain_openai import ChatOpenAI
from langchain.agents import create_tool_calling_agent, AgentExecutor
from langchain_core.prompts import ChatPromptTemplate

llm = ChatOpenAI(model="gpt-4o", temperature=0)
tools = [query_snowflake, get_table_schema, list_tables, detect_metric_anomaly]

prompt = ChatPromptTemplate.from_messages([
    ("system", """You are a data analyst with access to Snowflake data warehouse.

When answering data questions:
1. First list tables or get schema to understand available data
2. Generate precise SQL — always include a LIMIT clause
3. Interpret results in business terms, not just raw numbers
4. Compare to prior periods when possible for context
5. Flag data quality issues if you encounter NULL values or unexpected patterns

SQL guidelines:
- Use ILIKE for case-insensitive string matching
- Use DATEADD for date arithmetic
- Prefer CTEs (WITH clauses) for complex queries
- Always specify column names explicitly — avoid SELECT *"""),
    ("human", "{input}"),
    ("placeholder", "{agent_scratchpad}"),
])

agent = create_tool_calling_agent(llm, tools, prompt)
executor = AgentExecutor(agent=agent, tools=tools, verbose=True, max_iterations=8)

# Natural language query example
result = executor.invoke({
    "input": "What were our top 10 customers by revenue last month? Show their revenue and growth vs the month before."
})
print(result["output"])