Tutorial: Run a LangGraph agent over Snowflake data in a sandbox

This tutorial shows how to run a LangGraph StateGraph agent inside a Cortex Sandbox. The agent queries live Snowflake data using your injected session identity, calls SNOWFLAKE.CORTEX.COMPLETE for AI reasoning, and is contained by the platform’s egress baseline, so it cannot call an external AI provider or any other host outside that set.

When you finish, you will have a working pattern for deploying any LangGraph agent against Snowflake data with governance enforced and credentials never leaving the platform.

Why run LangGraph inside a sandbox

LangGraph agents are typically run in the same environment as the developer, which gives the agent access to the developer’s credentials, environment variables, and network. Running the agent inside a Cortex Sandbox changes three things:

  • Governance is enforced automatically. Every query the agent runs goes through your session identity. Row masking, column masking, and row access policies apply to what the agent can read, exactly as they would for a query you run directly. The agent cannot see more data than your role permits.
  • Egress is locked down. The sandbox’s outbound network access is controlled by an Egress policy. The default Egress() gives the container the platform baseline (Snowflake, its cloud-storage stages, and roughly 43 package-manager hosts) and nothing beyond that — so even if the agent tried to call an external LLM endpoint, the egress proxy would block it. Only SNOWFLAKE.CORTEX.COMPLETE (which runs as SQL inside Snowflake) is reachable.
  • No credentials enter the agent. The platform injects a session token at /snowflake/session/token. The agent script connects to Snowflake using only that token — no PAT, no environment variable, no connection string.

Time to complete: about 15 minutes.

Prerequisites

  • The snowflake-sandbox-python SDK installed and a Snowflake connection configured. See Cortex Sandboxes quickstart.
  • Your role must have access to SNOWFLAKE.ACCOUNT_USAGE.WAREHOUSE_METERING_HISTORY and USAGE on the Cortex COMPLETE function.

How the agent works

The agent is a LangGraph StateGraph with three nodes and one conditional edge:

question → query_node → (rows found?) → summarize_node → done
                              ↓ no
                          fallback_node → done
  • query_node runs SQL against WAREHOUSE_METERING_HISTORY, fetching the top five warehouses by credits used in the last seven days.
  • summarize_node passes the rows to SNOWFLAKE.CORTEX.COMPLETE and asks for a plain-English analysis.
  • fallback_node handles the case where no rows are returned (for example, no warehouse activity in the window).

Both query_node and summarize_node share a single Snowflake connection opened once at startup. The credential the platform provides is valid for the sandbox’s lifetime; opening multiple connections from it is not reliable.

Step 1: Write the agent script

Save the following as agent.py. This is the code that runs inside the sandbox.

# agent.py  — runs inside the sandbox
import json
import sys
import snowflake.connector
from typing import Any, TypedDict
from langgraph.graph import StateGraph, END

from snowflake.sandbox import get_snowflake_connection


# ---------------------------------------------------------------------------
# Single shared connection — opened once, reused across all graph nodes.
# get_snowflake_connection() resolves the session credential the platform
# provides; that credential is valid for the sandbox's lifetime, and opening
# multiple connections from it is not reliable.
# ---------------------------------------------------------------------------

def _make_connection() -> snowflake.connector.SnowflakeConnection:
    return get_snowflake_connection()


# ---------------------------------------------------------------------------
# Agent state
# ---------------------------------------------------------------------------

class AgentState(TypedDict):
    question: str
    rows: list
    answer: str


# ---------------------------------------------------------------------------
# Graph nodes
# ---------------------------------------------------------------------------

def query_node(state: AgentState, *, cur: snowflake.connector.cursor.SnowflakeCursor) -> AgentState:
    """Query WAREHOUSE_METERING_HISTORY for the last 7 days."""
    cur.execute("""
        SELECT warehouse_name,
               ROUND(SUM(credits_used), 2)  AS total_credits,
               COUNT(*)                     AS billing_periods
        FROM   snowflake.account_usage.warehouse_metering_history
        WHERE  start_time >= DATEADD('day', -7, CURRENT_TIMESTAMP())
        GROUP  BY 1
        ORDER  BY total_credits DESC
        LIMIT  5
    """)
    rows = [
        {"warehouse": str(r[0]), "credits": float(r[1]), "periods": int(r[2])}
        for r in cur.fetchall()
    ]
    return {**state, "rows": rows}


def summarize_node(state: AgentState, *, cur: snowflake.connector.cursor.SnowflakeCursor) -> AgentState:
    """Call SNOWFLAKE.CORTEX.COMPLETE with the query results."""
    prompt = (
        f"You are a Snowflake cost analyst. "
        f"A user asked: '{state['question']}'. "
        f"Here is 7-day warehouse credit usage (top 5 by credits): "
        + json.dumps(state["rows"])
        + " Write a 3-sentence answer that directly responds to the question, "
        "cites specific numbers, and notes any unusual patterns."
    )
    cur.execute("SELECT SNOWFLAKE.CORTEX.COMPLETE('llama3-8b', %s)", (prompt,))
    answer = (cur.fetchone() or ["No response from Cortex."])[0].strip()
    return {**state, "answer": answer}


def fallback_node(state: AgentState) -> AgentState:
    """Handle the case where no warehouse data was found."""
    return {**state, "answer": "No warehouse credit data was found for the last 7 days."}


# ---------------------------------------------------------------------------
# Routing function
# ---------------------------------------------------------------------------

def route_after_query(state: AgentState) -> str:
    return "summarize" if state["rows"] else "fallback"


# ---------------------------------------------------------------------------
# Build and run the graph
# ---------------------------------------------------------------------------

def build_graph(cur: snowflake.connector.cursor.SnowflakeCursor) -> Any:
    # Bind the shared cursor into the nodes via closures
    graph = StateGraph(AgentState)
    graph.add_node("query",    lambda s: query_node(s, cur=cur))
    graph.add_node("summarize", lambda s: summarize_node(s, cur=cur))
    graph.add_node("fallback",  fallback_node)

    graph.set_entry_point("query")
    graph.add_conditional_edges("query", route_after_query, {
        "summarize": "summarize",
        "fallback":  "fallback",
    })
    graph.add_edge("summarize", END)
    graph.add_edge("fallback",  END)
    return graph.compile()


if __name__ == "__main__":
    question = " ".join(sys.argv[1:]) or "Which warehouse used the most credits this week?"

    conn = _make_connection()
    cur = conn.cursor()
    try:
        app = build_graph(cur)
        result = app.invoke({"question": question, "rows": [], "answer": ""})
    finally:
        conn.close()

    print("Question:", result["question"])
    print()
    if result["rows"]:
        print("Data retrieved:")
        for r in result["rows"]:
            print(f"  {r['warehouse']}: {r['credits']} credits ({r['periods']} periods)")
        print()
    print("Answer:")
    print(result["answer"])

Replace llama3-8b with any Cortex model your role has access to. See COMPLETE (SNOWFLAKE.CORTEX) for the full model list.

Step 2: Write the driver

The driver starts the sandbox, sends agent.py into it, installs LangGraph inside it, then runs the agent. Save this as main.py in the same directory as agent.py.

# main.py  — runs on your machine (the driver)
from snowflake.sandbox import Image, Sandbox, Egress

QUESTION = "Which warehouse used the most credits this week, and is that unusual?"

print("Starting sandbox...")
sb = Sandbox.create(image=Image.from_catalog("sandbox-base"), egress=Egress())
try:
    print(f"Sandbox ready: {sb.id}")

    # Send the agent into the sandbox and install its dependencies there.
    # Public PyPI is in the platform egress baseline, so this needs no extra
    # egress configuration.
    sb.upload_file("agent.py", "/tmp/agent.py")
    sb.exec(["pip", "install", "--quiet", "langgraph", "langchain-core"])

    result = sb.exec(
        ["python", "/tmp/agent.py", QUESTION],
        timeout=120.0,
    )
    print(result.stdout)
    if result.stderr:
        print("--- stderr ---")
        print(result.stderr)
finally:
    sb.terminate()

Dependencies are installed into each sandbox rather than baked into a reusable image. pip reaches pypi.org and files.pythonhosted.org through the platform’s package-manager baseline, so the install needs no egress configuration and no external access integration.

Step 3: Run the example

Save agent.py and main.py in the same directory and run:

python main.py

The whole run takes roughly ten seconds: a few seconds to start the sandbox, a few more to install LangGraph, then the agent’s own query and Cortex call.

Expected output:

Starting sandbox...
Sandbox ready: cntr_...
Question: Which warehouse used the most credits this week, and is that unusual?

Data retrieved:
  COMPUTE_WH: 142.30 credits (504 billing periods)
  ANALYTICS_WH: 38.10 credits (168 billing periods)
  TRANSFORM_WH: 12.40 credits (44 billing periods)
  ...

Answer:
COMPUTE_WH was the top consumer this week at 142.3 credits across 504 billing
periods, which represents roughly four times the usage of the next warehouse.
This level of sustained activity suggests consistent scheduled workloads rather
than a spike. The remaining warehouses show normal ad-hoc and transformation
patterns with no unusual outliers.

The exact output depends on your account’s warehouse activity.

What this demonstrates

  • No credentials in the agent. agent.py connects using only the token the platform binds at /snowflake/session/token. Your PAT is not passed to the container.
  • One connection for the graph. The agent opens a single snowflake.connector connection and shares it across all graph nodes via closures. The platform session token is reliable for the sandbox’s lifetime but not for multiple simultaneous connections.
  • Governance is enforced. The agent’s queries run under your session identity. If your role has column masking or row access policies on WAREHOUSE_METERING_HISTORY, they apply — the agent cannot read past them.
  • External LLM endpoints are blocked. The platform baseline Egress() does not include external AI provider endpoints. The agent cannot call OpenAI, Anthropic, or any other API even if LangGraph is configured to do so. Cortex AI is available through SQL, not HTTP, so it is unaffected.
  • LangGraph runs unmodified. The agent.py script uses the standard LangGraph API — StateGraph, add_node, add_conditional_edges. No sandbox-specific changes to the agent code are needed.
  • Dependencies install from inside the sandbox. pypi.org and files.pythonhosted.org are in the platform’s package-manager egress baseline, so installing LangGraph needs no egress entry and no external access integration.

Going further

  • Add more graph nodes. The StateGraph can include nodes that compare against a historical baseline, check for anomalies across multiple time windows, or generate a formatted report.
  • Fan out across questions. Use ThreadPoolExecutor to launch multiple sandboxes in parallel, each answering a different question or analyzing a different table — the same pattern as Tutorial: Run Cortex batch inference over a Snowflake table.
  • Add a Snowflake secret for an external LLM. To use an external model instead of Cortex, mount a Secret and add the model’s host to Egress. The credential’s value is not placed in the container. See Tutorial: Give a sandbox secure access to an external API.
  • Use a warm sandbox for interactive sessions. Rather than creating a sandbox per question, keep one alive and call sb.send(question) in a loop to keep the agent warm across multiple questions without restarting.
Next
Prerequisites