The real problem
Why this is harder than it looks
Snowflake's SQL API is clean and well-structured. The OAuth 2.0 flow is standard. Most developers figure the auth layer will take a day. Then they hit what the documentation buries in footnotes.
Snowflake OAuth requires you to create an OAuth Security Integration inside your Snowflake account using SQL — there's no UI-driven app registration. Each customer's Snowflake instance is its own account with a unique domain (orgname-accountname.snowflakecomputing.com), and the OAuth endpoints and API base URLs are derived from that account locator. That means you can't hardcode a base URL. Every user in your product is connecting to a different Snowflake account, and your code must resolve the correct endpoint per user at runtime. Scalekit handles this: the account domain is captured during the OAuth flow and applied automatically on every subsequent API call.
Then there's the role and warehouse context problem. Snowflake's SQL API requires an active warehouse to execute queries — if none is specified and the user's default warehouse is suspended or unset, queries fail silently or return an unhelpful error. Roles in Snowflake are additive and context-sensitive: a user may have multiple roles with different data access, and the role active at query time determines what they can see. Getting this wrong means queries that work in development fail in production when a different user with a different default role connects.
On top of that: per-user token isolation across tenants, proactive refresh before expiry, revocation detection when a user's Snowflake session is revoked by an admin, and the BYOC requirement that means you must create and manage an OAuth Security Integration in every customer's Snowflake account. Scalekit handles the token lifecycle end-to-end. Your agent names a tool and passes SQL.
Capabilities
What your agent can do with Snowflake
Once connected, your agent has 14 pre-built tools covering query execution, schema introspection, and warehouse management:
- Execute SQL against any database: run queries synchronously or asynchronously, with role, warehouse, database, and schema context per call
- Inspect schemas and tables: browse databases, schemas, tables, and column metadata via INFORMATION_SCHEMA — without writing boilerplate SQL
- Track async query status: submit long-running queries asynchronously, poll status by handle, and retrieve paginated result partitions
- Understand data relationships: query table constraints, primary keys, referential constraints, and imported/exported key relationships
- Audit access and capacity: show grants, list warehouses, and inspect role-level permissions programmatically
Setup context
What we're building
This guide connects a data assistant agent to Snowflake — helping analysts and engineers query warehouse data, explore schemas, and run SQL without leaving your product.
🤖
Example agent
Data assistant executing SQL and exploring schemas on behalf of each analyst
🔐
Auth model
B2B SaaS — each user connects their own Snowflake account. identifier = your user ID
Setup
1 Setup: One SDK, One credential
Install the Scalekit SDK. The only credential your application manages is the Scalekit API key — no Snowflake secrets, no user tokens, nothing belonging to your customers.
pip install scalekit-sdk-python
npm install @scalekit-sdk/node
import scalekit.client
import os
from dotenv import load_dotenv
load_dotenv()
scalekit = scalekit.client.ScalekitClient(
client_id=os.getenv("SCALEKIT_CLIENT_ID"),
client_secret=os.getenv("SCALEKIT_CLIENT_SECRET"),
env_url=os.getenv("SCALEKIT_ENV_URL"),
)
actions = scalekit.actions
import { ScalekitClient } from '@scalekit-sdk/node';
import 'dotenv/config';
const scalekit = new ScalekitClient(
process.env.SCALEKIT_ENV_URL,
process.env.SCALEKIT_CLIENT_ID,
process.env.SCALEKIT_CLIENT_SECRET
);
const actions = scalekit.actions;
Already have credentials?
Connected Accounts
2 Per-User Auth: Creating connected accounts
Each analyst gets their own Connected Account, giving them a dedicated auth context. The identifier is any unique string from your system — a UUID, email, whatever you use internally.
response = actions.get_or_create_connected_account(
connection_name="snowflake",
identifier="user_sf_456" # your internal user ID
)
connected_account = response.connected_account
print(f"Status: {connected_account.status}")
# Status: PENDING — user hasn't authorized yet
const response = await actions.getOrCreateConnectedAccount({
connectionName: "snowflake",
identifier: "user_sf_456" // your internal user ID
});
const connectedAccount = response.connectedAccount;
console.log(`Status: ${connectedAccount.status}`);
// Status: PENDING — user hasn't authorized yet
This call is idempotent — safe to call on every session start. Returns the existing account if one already exists.
Authorization Flow
3 The authorization flow
The user authorizes your agent once. Scalekit generates the OAuth URL with correct scopes, PKCE challenge, and redirect handling pre-configured. After approval, you never see the token.
if connected_account.status != "ACTIVE":
link = actions.get_authorization_link(
connection_name="snowflake",
identifier="user_sf_456"
)
# Redirect user → Snowflake OAuth consent screen
# Scalekit captures the token and account domain on callback
return redirect(link.link)
if (connectedAccount.status !== "ACTIVE") {
const { link } = await actions.getAuthorizationLink({
connectionName: "snowflake",
identifier: "user_sf_456"
});
// Redirect user → Snowflake OAuth consent screen
// Scalekit captures the token and account domain on callback
return redirect(link);
}
Token management is automatic
After the user approves, Scalekit stores encrypted tokens and the connected account moves to ACTIVE. Access
tokens refresh before expiry. If a user's token is revoked by a Snowflake admin, the account moves to REVOKED
— no silent failures. Check account.status before critical operations.
BYOC is required for Snowflake
Snowflake requires you to create an OAuth Security Integration in your own Snowflake account using SQL.
Run the CREATE SECURITY INTEGRATION statement in Snowsight, retrieve the Client ID and Secret via
SYSTEM$SHOW_OAUTH_CLIENT_SECRETS, then paste them into the Scalekit dashboard. Token management is
fully handled after that one-time setup.
Calling Snowflake
4 Calling Snowflake: What your agent writes
With the connected account active, your agent calls Snowflake actions via execute_tool. Name the tool, pass parameters. Scalekit handles token retrieval, account domain resolution, and request construction.
Execute a SQL query
Run any SQL statement against a specified database and schema. Pass warehouse and role per call to override the user's defaults — useful when your agent needs to query across different contexts.
result = actions.execute_tool(
identifier="user_sf_456",
tool_name="snowflake_execute_query",
tool_input={
"statement": "SELECT order_id, customer_id, total_amount, status FROM orders WHERE status = 'PENDING' ORDER BY total_amount DESC LIMIT 20",
"database": "ANALYTICS",
"schema": "PUBLIC",
"warehouse": "COMPUTE_WH",
"role": "ANALYST"
}
)
# Returns: { "statementHandle": "...", "data": [[...], ...], "resultSetMetaData": {...} }
const result = await actions.executeTool({
identifier: "user_sf_456",
toolName: "snowflake_execute_query",
toolInput: {
statement: "SELECT order_id, customer_id, total_amount, status FROM orders WHERE status = 'PENDING' ORDER BY total_amount DESC LIMIT 20",
database: "ANALYTICS",
schema: "PUBLIC",
warehouse: "COMPUTE_WH",
role: "ANALYST"
}
});
// Returns: { statementHandle: "...", data: [[...], ...], resultSetMetaData: {...} }
Explore schema and tables
Enumerate available schemas and tables before running queries — useful for agents that need to discover what data exists in a database before constructing SQL.
# List schemas in a database
schemata = actions.execute_tool(
identifier="user_sf_456",
tool_name="snowflake_get_schemata",
tool_input={
"database": "ANALYTICS",
"warehouse": "COMPUTE_WH"
}
)
# List tables in a schema
tables = actions.execute_tool(
identifier="user_sf_456",
tool_name="snowflake_get_tables",
tool_input={
"database": "ANALYTICS",
"schema": "PUBLIC",
"warehouse": "COMPUTE_WH",
"limit": 50
}
)
# Returns table names, types, row counts, and creation timestamps
// List schemas in a database
const schemata = await actions.executeTool({
identifier: "user_sf_456",
toolName: "snowflake_get_schemata",
toolInput: {
database: "ANALYTICS",
warehouse: "COMPUTE_WH"
}
});
// List tables in a schema
const tables = await actions.executeTool({
identifier: "user_sf_456",
toolName: "snowflake_get_tables",
toolInput: {
database: "ANALYTICS",
schema: "PUBLIC",
warehouse: "COMPUTE_WH",
limit: 50
}
});
// Returns table names, types, row counts, and creation timestamps
Inspect column metadata
Fetch column definitions for a specific table — data types, nullability, and defaults. Useful for agents that generate SQL dynamically and need to know the schema before constructing a query.
columns = actions.execute_tool(
identifier="user_sf_456",
tool_name="snowflake_get_columns",
tool_input={
"database": "ANALYTICS",
"schema": "PUBLIC",
"table": "ORDERS",
"warehouse": "COMPUTE_WH"
}
)
# Returns: column names, data types, is_nullable, column_default
const columns = await actions.executeTool({
identifier: "user_sf_456",
toolName: "snowflake_get_columns",
toolInput: {
database: "ANALYTICS",
schema: "PUBLIC",
table: "ORDERS",
warehouse: "COMPUTE_WH"
}
});
// Returns: column names, data types, is_nullable, column_default
Run async queries and poll results
For long-running queries, submit asynchronously and poll by handle. Use snowflake_get_query_partition to page through large result sets without loading everything at once.
import time
# Submit async
submit = actions.execute_tool(
identifier="user_sf_456",
tool_name="snowflake_execute_query",
tool_input={
"statement": "SELECT * FROM large_events_table WHERE event_date >= '2026-01-01'",
"database": "ANALYTICS",
"schema": "PUBLIC",
"warehouse": "COMPUTE_WH",
"async": True
}
)
handle = submit["statementHandle"]
# Poll until complete
while True:
status = actions.execute_tool(
identifier="user_sf_456",
tool_name="snowflake_get_query_status",
tool_input={"statement_handle": handle}
)
if status["status"] == "success":
break
time.sleep(2)
# Fetch first partition
results = actions.execute_tool(
identifier="user_sf_456",
tool_name="snowflake_get_query_partition",
tool_input={"statement_handle": handle, "partition": 0}
)
// Submit async
const submit = await actions.executeTool({
identifier: "user_sf_456",
toolName: "snowflake_execute_query",
toolInput: {
statement: "SELECT * FROM large_events_table WHERE event_date >= '2026-01-01'",
database: "ANALYTICS",
schema: "PUBLIC",
warehouse: "COMPUTE_WH",
async: true
}
});
const handle = submit.statementHandle;
// Poll until complete
let status;
do {
await new Promise(r => setTimeout(r, 2000));
status = await actions.executeTool({
identifier: "user_sf_456",
toolName: "snowflake_get_query_status",
toolInput: { statement_handle: handle }
});
} while (status.status !== "success");
// Fetch first partition
const results = await actions.executeTool({
identifier: "user_sf_456",
toolName: "snowflake_get_query_partition",
toolInput: { statement_handle: handle, partition: 0 }
});
Framework wiring
5 Wiring into your agent framework
Scalekit integrates directly with LangChain. The agent decides what to query; Scalekit handles auth on every invocation. No token plumbing in your agent logic.
from langchain_anthropic import ChatAnthropic
from langchain.agents import AgentExecutor, create_tool_calling_agent
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from scalekit.langchain import get_tools
sf_tools = get_tools(
connection_name="snowflake",
identifier="user_sf_456"
)
prompt = ChatPromptTemplate.from_messages([
("system", "You are a data assistant. Use the available tools to help query and explore Snowflake data warehouses."),
MessagesPlaceholder("chat_history", optional=True),
("human", "{input}"),
MessagesPlaceholder("agent_scratchpad"),
])
agent = create_tool_calling_agent(ChatAnthropic(model="claude-sonnet-4-6"), sf_tools, prompt)
result = AgentExecutor(agent=agent, tools=sf_tools).invoke({
"input": "Show me the top 10 customers by revenue this quarter from the ANALYTICS database"
})
import { ChatAnthropic } from "@langchain/anthropic";
import { AgentExecutor, createToolCallingAgent } from "langchain/agents";
import { ChatPromptTemplate, MessagesPlaceholder } from "@langchain/core/prompts";
import { getTools } from "@scalekit-sdk/langchain";
const sfTools = getTools({
connectionName: "snowflake",
identifier: "user_sf_456"
});
const prompt = ChatPromptTemplate.fromMessages([
["system", "You are a data assistant. Use the available tools to help query and explore Snowflake data warehouses."],
new MessagesPlaceholder("chat_history", true),
["human", "{input}"],
new MessagesPlaceholder("agent_scratchpad"),
]);
const agent = await createToolCallingAgent({
llm: new ChatAnthropic({ model: "claude-sonnet-4-6" }),
tools: sfTools,
prompt
});
const result = await AgentExecutor.fromAgentAndTools({
agent,
tools: sfTools
}).invoke({
input: "Show me the top 10 customers by revenue this quarter from the ANALYTICS database"
});
Other frameworks supported
Tool reference
All 14 Snowflake tools
Grouped by capability. Your agent calls tools by name — no API wrappers to write.
Execute one or more SQL statements. Supports sync and async modes, role/warehouse/database context, bind variables, and idempotent retries via request_id
Cancel a running statement by its statement handle
snowflake_get_query_status
Get the execution status and first partition metadata for an async statement by handle
snowflake_get_query_partition
Retrieve a specific result partition (0-based index) for a completed async statement
snowflake_show_databases_schemas
Run SHOW DATABASES or SHOW SCHEMAS with optional LIKE pattern and database scope
Query INFORMATION_SCHEMA.SCHEMATA for schema metadata within a database
Query INFORMATION_SCHEMA.TABLES with optional schema filter, table name pattern, and row limit
Query INFORMATION_SCHEMA.COLUMNS for column names, data types, nullability, and defaults — with optional table, schema, and column name filters
snowflake_get_table_constraints
Query INFORMATION_SCHEMA.TABLE_CONSTRAINTS with optional constraint type, schema, and table filters
snowflake_get_referential_constraints
Query INFORMATION_SCHEMA.REFERENTIAL_CONSTRAINTS for foreign key relationships between tables
snowflake_show_primary_keys
Run SHOW PRIMARY KEYS with optional database, schema, and table scope
snowflake_show_imported_exported_keys
Run SHOW IMPORTED KEYS or SHOW EXPORTED KEYS for a table. Use fully-qualified scope (database + schema + table) for reliable execution
snowflake_show_warehouses
Run SHOW WAREHOUSES to list available compute warehouses with optional LIKE pattern filter
Run SHOW GRANTS in common modes — to role, to user, of role, or on a specific object — to inspect role-level permissions
Connector notes
Snowflake-specific behavior
BYOC is required — no managed app available
Snowflake does not have a central app registry. You must create an OAuth Security Integration inside your
own Snowflake account using SQL (CREATE SECURITY INTEGRATION), retrieve the Client ID and Secret via
SYSTEM$SHOW_OAUTH_CLIENT_SECRETS, and paste them into the Scalekit dashboard. Each customer's Snowflake
account is a separate OAuth registration.
Always specify warehouse — absent defaults cause silent failures
Snowflake requires an active virtual warehouse to execute queries. If the user's default warehouse is
suspended or unset, queries fail with an ambiguous error. Pass warehouse explicitly on every
snowflake_execute_query call rather than relying on session defaults.
Account domain is resolved automatically
Every Snowflake account has a unique domain (orgname-accountname.snowflakecomputing.com). Scalekit captures
this during the OAuth flow and routes all subsequent API calls to the correct endpoint automatically.
You do not need to store or pass the account locator in your application code.
Infrastructure decision
Why not build this yourself
The Snowflake OAuth flow is documented. Token storage isn't technically hard. But here's what you're actually signing up for:
PROBLEM 01
Each customer's Snowflake account has a unique domain — you must resolve and store the correct API endpoint per user, and route every call to the right account at runtime
PROBLEM 02
BYOC is mandatory — OAuth Security Integrations must be created via SQL inside each Snowflake account, there's no central app registry or UI-driven setup path
PROBLEM 03
Per-user token isolation across a multi-tenant system — one analyst's Snowflake credentials must never be accessible to another, even within the same organization
PROBLEM 04
Revocation detection when a Snowflake admin revokes a user's OAuth token — and graceful handling so the agent stops making calls and surfaces a re-authorization prompt
That's one connector. Your agent product will eventually need Salesforce, Gmail, HubSpot, Notion, and whatever else your customers ask for. Each has its own OAuth quirks and failure modes.
Scalekit maintains every connector. You maintain none of them.
