Cortex Sandboxes SDK reference¶

This page is the reference for the snowflake.sandbox Python API. To install the SDK and configure a Snowflake connection, see the Cortex Sandboxes quickstart.

Common operations¶

Everything above is documented in full below. For the errors these calls raise, see the error reference.

Public surface¶

Copy-paste form:

from snowflake.sandbox import (
    Sandbox, ExecResult, RunResult, StreamLine, ExecStream,
    App, Function, Image, Secret, Egress, Bundle, StageMount, McpServer,
    broker_env_var, session_loop, Config, using, get_snowflake_connection, current_config, close_connections,
    AsyncSandbox, SandboxAuthError, SandboxBuildError, SandboxConflictError, SandboxContractWarning, SandboxCreateError, SandboxError, SandboxExecError, SandboxExecTimeoutError, SandboxFailure, SandboxFileTooLargeError, SandboxNotFoundError, SandboxNotImplementedError, SandboxNotReadyError, SandboxRateLimitError, SandboxTransportError, SandboxValidationError,
    ConnectionLike, DeployPlan, DeployResult, FileInfo, FileWatchEvent, FileWatchEventType, Job, McpConnectionStatus, McpServerInfo, Session, SessionApp, Shell, ShellClosed, SyncExecStream, SyncJob, SyncShell, agent_session, agent_session_sync, deploy_async, deploy_async_sync, destroy_sandbox, destroy_sandbox_async, get_sandbox, get_sandbox_async, get_sandbox_by_name, get_sandbox_by_name_async, list_mcp_servers, list_sandboxes, list_sandboxes_async, mcp_server_status, mcp_token_env_var, resolve_config, shutdown, validate_mcp_servers,
)

RunResult is the outcome type of a detached run (create(command=...) + wait).

The Sandbox object¶

Sandbox represents one container. Construction is lazy—no network call happens until you call create(), or enter the object as a with context manager.

Create¶

Sandbox.create is the one-call factory. It creates the sandbox and returns a live handle.

sb = Sandbox.create(
    memory="4g",
    egress=Egress(),
    name="my-agent",
)

Key parameters of Sandbox.create() (all keyword-only):

Use create() in a with block to destroy the container on exit:

with Sandbox.create() as sb:
    result = sb.exec(["python", "-c", "print(40 + 2)"])

The earlier sandbox.toml manifest format is retired. Use Sandbox.create() (with code=) or the App interface instead.

Properties¶

Sandbox methods¶

Create and connect¶

Sandbox.connect()¶

sb = Sandbox.connect("my-agent")
result = sb.exec(["python", "-c", "print('still running')"])

Sandbox.list()¶

sandboxes = Sandbox.list()
for sb in sandboxes:
    print(sb.id, sb.status)

# What MY_ROLE can see. Resolved from the ambient connection at that role;
# under a bare env token, which carries its own role, role= only applies the
# server-side ?role= filter.
mine = Sandbox.list(role="MY_ROLE")

Run commands¶

sb.exec()¶

result = sb.exec(["python", "-c", "print(40 + 2)"])
print(result.stdout)   # "42\n"

sb.exec_stream()¶

for line in sb.exec_stream(["python", "generate_report.py"]):
    print(line.stream, line.data)

Wait and send¶

sb.wait_until_ready()¶

# Fire all creates immediately
sandboxes = [Sandbox.create(wait=False)
             for _ in range(10)]
# Wait for all in parallel
with ThreadPoolExecutor() as pool:
    pool.map(lambda sb: sb.wait_until_ready(), sandboxes)

sb.wait()¶

sb = Sandbox.create(
    command=["python", "main.py"],
)
result = sb.wait()
print(result.status, result.exit_code)

sb.send()¶

sb = Sandbox.create(
    command=["python", "daemon.py"],
)
reply = sb.send("what changed in prod yesterday?")
print(reply)

Manage the container¶

sb.refresh()¶

status = sb.refresh()
print(status)  # "ready", "pending", "dead", "failed", or "unknown"

sb.refresh_live_status()¶

status = sb.refresh_live_status()
print(status)  # the LIVE process status, not the create-time snapshot

sb.logs()¶

output = sb.logs(tail=100)
print(output)

# What the commands you ran with `exec()` printed.
print(sb.logs(source="exec"))

sb.shell()¶

with sb.shell() as sh:
    sh.send("ls -la\n")
    for chunk in sh.output():
        sys.stdout.buffer.write(chunk)

sb.suspend()¶

sb.exec(["sh", "-c", "echo work > /var/tmp/state"])
sb.suspend()                     # state is banked
print(sb.exec(["cat", "/var/tmp/state"]).stdout)   # resumes; prints "work"

sb.resume()¶

sb.resume()    # returns before the sandbox is serving; the next exec waits

sb.poll()¶

while (code := sb.poll()) is None:
    time.sleep(1)
print("exit code:", code)

sb.get_tags()¶

tags = sb.get_tags()
print(tags.get("team"))

sb.set_tags()¶

with Sandbox.create(tags={"team": "sandbox", "env": "dev"}) as sb:
    sb.exec(["python", "main.py"])

File operations¶

sb.upload_file()¶

sb.upload_file("./data.csv", "/app/data.csv")

sb.upload_dir()¶

plan = sb.upload_dir("./project", "/app", exclude=["*.csv"])
print(f"sent {plan.file_count} files, skipped {plan.skipped_by_reason()}")

sb.download_file()¶

sb.download_file("/app/output.csv", "./output.csv")

sb.read_text()¶

content = sb.read_text("/app/output.txt")

sb.read_bytes()¶

data = sb.read_bytes("/app/output.bin")

sb.write_text()¶

sb.write_text("/app/config.json", '{"key": "value"}')

sb.write_bytes()¶

sb.write_bytes("/app/model.pkl", model_bytes)

sb.list_files()¶

entries = sb.list_files("/app")
for entry in entries:
    print(entry.path, entry.size_bytes, entry.is_dir)

sb.make_directory()¶

sb.make_directory("/app/output", create_parents=True)

sb.stat()¶

info = sb.stat("/app/output.txt")
print(info.size_bytes)

sb.remove()¶

sb.remove("/app/temp", recursive=True)

sb.stage_put()¶

# This raises SandboxNotImplementedError. Mount the stage and write
# through the mount instead:
data = StageMount.from_stage("MY_STAGE", mount_path="/data")
with Sandbox.create(stage_mounts=[data]) as sb:
    sb.write_bytes("/data/results.parquet", payload)

sb.stage_get()¶

# This raises SandboxNotImplementedError. Mount the stage and read
# through the mount instead:
data = StageMount.from_stage("MY_STAGE", mount_path="/data", readonly=True)
with Sandbox.create(stage_mounts=[data]) as sb:
    model = sb.read_bytes("/data/model.pkl")

sb.watch()¶

for event in sb.watch("/app"):
    print(event.type, event.paths)

Result and stream types¶

Sandbox.exec() returns an ExecResult, a named tuple with the following fields:

Sandbox.exec_stream() yields StreamLine named tuples:

The SyncExecStream returned by exec_stream() exposes exit_code, elapsed_ms, and error after the iterator drains. It does not raise on a non-zero exit; check stream.exit_code. It raises SandboxExecError on an error frame (meaning the exec never produced a result).

Sandbox.wait() returns a RunResult:

File and watch types¶

FileInfo – Metadata for one path, returned by Sandbox.list_files() / .stat(). Fields: path, is_dir, size_bytes, modified_at.

Mirrors Modal’s Sandbox.filesystem.stat shape. Both producers are implemented (files.py runs an in-container python3 helper over exec); modified_at is a Unix epoch (seconds, UTC) or None when the server did not report it.

FileWatchEvent – One filesystem-change event yielded by Sandbox.watch(). Fields: paths, type.

Mirrors Modal’s FileWatchEvent: paths are the absolute in-sandbox path(s) the change concerns (usually one), and type is the coarse FileWatchEventType.

FileWatchEventType – Coarse change class for a Sandbox.watch() event. Members: Unknown, Access, Create, Modify, Delete.

Mirrors Modal’s FileWatchEventType. The sandbox observes raw Linux inotify(7) masks and maps them onto this small enum — several inotify bits collapse to one class (e.g. IN_MODIFY and IN_CLOSE_WRITE both read as Modify; IN_CREATE and IN_MOVED_TO both as Create). A mask that maps to nothing meaningful surfaces as Unknown rather than being dropped, so a contract drift is visible instead of silent. str-valued so it compares equal to its wire string and JSON-serializes cleanly.

Images, secrets, egress, and bundles¶

The declarative specs (Image, Secret, Egress, Bundle, StageMount, DeploySpec) describe a deployable app. @app.function(...) takes them directly, and they compile to a DeploySpec. There is no manifest file.

Image¶

An Image is a reference to a catalog base image. Use Image.from_catalog() to select one:

image = Image.from_catalog("sandbox-base")

Pass an Image (or a plain catalog-name string) directly to Sandbox.create():

with Sandbox.create(image=Image.from_catalog("sandbox-base")) as sb:
    result = sb.exec(["python", "-c", "print(40 + 2)"])

SDK-side image dependency baking (the layered-image builders pip_install, run_commands, copy_local_*, env, workdir, and build) is temporarily unavailable and will return in a later release. Until then, either use a base image that already includes the dependencies you need, or install them into the running sandbox with sb.exec(["pip", "install", ...]). Public PyPI is in the always-on egress baseline, so a runtime install needs no egress configuration and no external access integration.

Secret¶

Secret is how a sandbox uses an external credential. It references a Snowflake SECRET object. The secret’s value is not placed in the container; the platform supplies it on outbound requests to the declared host. See Cortex Sandboxes concepts and security.

secret = Secret.from_name("<your_database>.<your_schema>.GH_TOKEN", env_var="GH_TOKEN", host="github.com")

• Secret.from_name(fqn, *, env_var, host, hosts) references a SECRET by fully qualified name. host and hosts are mutually exclusive.
• The env var that receives the value is env_var if set, otherwise derived from the fully qualified name via broker_env_var.

Egress¶

Egress controls outbound network access. There are exactly three levels, and the middle one is the default:

The baseline is not the internet. It is a curated host set that the platform maintains, and anything outside it is refused. A host you need that is not on that list requires an external access integration at every level, including the baseline.

egress = Egress()                                # baseline
egress = Egress.only()                           # closed: nothing arbitrary
egress = Egress(allow_public_package_registries=False, external_access_integrations=("MY_EAI",))

Key fields:

broker_env_var(fqn, *, key=None) returns the environment variable name that the secret broker uses for a fully qualified SECRET name. Pass key="password" or "username" for a PASSWORD-type secret.

Bundle¶

A Bundle is a source-tree slice uploaded to the sandbox.

bundle = Bundle.from_dir(".", include=["tools/jira-triage/**"])

Bundle.from_dir(root, *, include, exclude) constructs a bundle from a root directory, where include and exclude are glob patterns.

StageMount¶

A StageMount mounts a Snowflake stage or workspace into the container.

mount = StageMount.from_stage("MY_STAGE", mount_path="/data")
ws_mount = StageMount.from_workspace("my_workspace", mount_path="/ws", readonly=True)

• StageMount.from_stage(name, *, mount_path, readonly) mounts a regular Snowflake stage.
• StageMount.from_workspace(name, *, mount_path, readonly) mounts a Snowflake workspace.

McpServer¶

An McpServer declares an EXTERNAL MCP SERVER the sandbox may call. Addressing is by fully-qualified name, the same form Cortex Agents take; Snowflake resolves the endpoint, the API integration, and the calling user’s OAuth secret server-side. A bare FQN string works wherever an McpServer does.

with Sandbox.create(mcp_servers=["<your_database>.<your_schema>.GITHUB_MCP_SERVER"]) as sb:
    ...

gh = McpServer.external("<your_database>.<your_schema>.GITHUB_MCP_SERVER", name="github")

• McpServer.external(fqn, *, name, profile) declares a server. name keys mcp.json and is uppercased into the dummy-token env var (SANDBOX_MCP_TOKEN_GITHUB); it defaults to the FQN’s last part.
• list_mcp_servers() lists the account’s external MCP servers with your per-user authorization status (CONNECTED, NOT_CONNECTED, NEEDS_REAUTH, INTEGRATION_NOT_FOUND, UNKNOWN). Only CONNECTED servers reach a sandbox.
• The OAuth token is not placed in the container. The platform supplies it on requests to the server’s host, the same way it handles a Secret.

DeploySpec¶

DeploySpec is the internal representation that the decorator API and the CLI compile to. It records the image, entry command, memory, CPU, environment, egress, secrets, bundle, and stage mounts for a deployment. You typically do not construct one directly.

The App interface¶

The App decorator layer lets you call a typed Python function in a sandbox with .remote(), or run it in-process with .local().

from snowflake.sandbox import App, Image, Secret, Egress

app = App("my-agent")

@app.function(
    image=Image.from_catalog("sandbox-base"),
    entry=["python3", "agent.py"],
    secrets=[Secret.from_name("<your_database>.<your_schema>.GH_TOKEN", env_var="GH_TOKEN", host="github.com")],
    egress=Egress(),
)
def agent() -> None: ...

A registered Function supports:

A function can run in one of two modes:

• Process mode: entry= is specified. fn.remote(**kwargs) passes keyword arguments as environment variable overrides, and the function body runs only when you call fn() or fn.local() directly.
• Function mode: no entry=. The function body runs in the container; fn.remote(**kwargs) JSON-encodes the keyword arguments and runs the body inside the sandbox.

You can also register a stateful, keep-alive class with @app.session(...). The class defines on_message(self, msg) for each turn and may use the @enter decorator to mark a cold-start hook. The resulting SessionApp.session(key=...) deploys the daemon and returns a Session you drive with s.send(msg).

session_loop(handle) is the in-container half of a warm agent. handle(msg, state) is called for each incoming message; state is a plain dict that persists across calls in process memory — for as long as the process lives. It runs forever. Note that an idle auto-suspend deletes the container rather than checkpointing its memory, so state does not survive one and the loop is not restarted; keep anything you cannot recompute on the persistent rootfs (under /var/tmp), not in state and not in /tmp.

Configuration and connections¶

Name the connection where you construct the thing. connection= is on every entry point that does not already have a handle to carry it — Sandbox.create(), Sandbox.connect(), Sandbox.list(), list_sandboxes(), get_sandbox_by_name(), destroy_sandbox(), Job.from_id(), deploy_async(), snowflake.sandbox.deploy.deploy_spec(), Session.from_id(), agent_session() — and on both the sync and async twins:

from snowflake.sandbox import Sandbox, destroy_sandbox, list_sandboxes

sb = Sandbox.create(image="python-3.11", connection="prod_account")
sb.exec(["echo", "hi"])                                  # the handle carries it
for info in list_sandboxes(connection="prod_account"):  # no handle yet, so name it
    print(info.name)
destroy_sandbox(sb.id, connection="prod_account")

It takes one of four things (typed as ConnectionLike), mirroring the connector’s connect(connection_name=...): a connection name from ~/.snowflake/connections.toml, a live snowflake.connector connection you already opened, a Snowpark Session — what you hold inside Streamlit-in-Snowflake and stored procedures — or a Config of raw credentials for a process that has no connections file at all:

from snowflake.sandbox import Config, Sandbox

sb = Sandbox.create(connection=Config(account="my_account", pat=my_token))

# Reuse a session you already hold rather than authenticating twice:
from snowflake.snowpark.context import get_active_session
sb = Sandbox.create(connection=get_active_session())

A connection or Session you opened stays yours to close — the SDK reuses it and never closes what it did not open.

Resolution is memoised per connection and role, so passing it on every call costs one login, not one per call. Every handle returned carries the connection onward, so you name it once per handle rather than once per operation.

role= overrides the role the session uses when creating sandboxes, and is applied when the session is minted — so it composes with a connection name only. A live connection or Snowpark Session is already logged in, and raw credentials carry their own role, so passing role= with any of those is refused rather than silently ignored.

A role that cannot be applied is an error on every path — create(), and the read paths (list_sandboxes(), get_sandbox_by_name(), Sandbox.connect(), destroy_sandbox()) which used to tolerate it because role= there also carried the server-side ?role= filter, valid under any credential. role= means one thing now: the credential. The filter is still applied, as an internal narrowing consistent with that credential rather than a second meaning you can reach independently. If omitted, the role comes from the connection. A role the user does not hold is refused at login.

role is not an isolation boundary — user + account is. Listing with no role= returns every sandbox you own across all your roles, so a sandbox created under one role is still yours to see, attach to and terminate from a session at another. Passing role= narrows the server-side filter; it does not partition ownership.

using() is the scoped form, for when a sequence of calls shares one connection and repeating the argument would be noise. It binds a connection to a block:

from snowflake.sandbox import Sandbox, list_sandboxes, using

with using("prod_account"):
    for info in list_sandboxes():      # reads prod_account
        print(info.name)
    sb = Sandbox.connect("my-agent")   # attaches over prod_account

It takes the same arguments as connection= plus the client knobs, and yields the resolved config. Everything inside the block uses that connection. Nothing outside it is affected: the binding lives in a contextvars.ContextVar, so it cannot outlive the block, and two threads or asyncio tasks can hold two different connections at the same time. A sandbox handle built outside a block keeps its own connection when used inside one — using() decides where a newly acquired transport comes from, and never re-points an existing one. A per-call role= composes, minted from the block’s connection; repeating the same role is a no-op, while a different one raises SandboxError rather than silently preferring either.

Name nothing and the SDK resolves an ambient connection, the same way the snow CLI does: a credential in the environment (SNOWFLAKE_PAT / SNOWFLAKE_TOKEN, which is how a process running inside a sandbox authenticates with no config at all), else SNOWFLAKE_DEFAULT_CONNECTION_NAME or default_connection_name in ~/.snowflake/config.toml.

Those two are the process-wide default, and the SDK deliberately offers no setter of its own for it:

import os
os.environ["SNOWFLAKE_DEFAULT_CONNECTION_NAME"] = "prod_account"

# ~/.snowflake/config.toml
default_connection_name = "prod_account"

One source of truth, honoured by every Snowflake library in the process — the connector, the snow CLI, Snowpark, Streamlit — so your sandboxes and your cursors cannot drift onto different accounts. A sandbox-only global would carry the same value with a different blast radius, and a caller using the SDK alongside snowflake.connector could end up running SQL against prod and sandboxes against dev with nothing warning them.

It is read when a connection is resolved, not at import, so setting it works mid-process — including from a notebook’s first cell, which a using() block cannot span.

current_config() returns the resolved config — useful for checking which account, host, and role the SDK picked. It is a reader: it never changes what a later call resolves, and it does not report an enclosing using() block (use the value the block yields for that).

close_connections() releases everything the process resolved — the connector sessions, the resolution memo, and every connection-bound transport. It is a teardown, not an un-configure: the next call resolves again and works.

get_snowflake_connection(**kwargs) returns a Snowflake connection for code running inside a Cortex Sandbox. Prefer it for in-sandbox code, production included. It resolves the credential the way the platform provides it, in order: SNOWFLAKE_PAT or SNOWFLAKE_TOKEN in the environment; a token file named by SNOWFLAKE_TOKEN_FILE_PATH, else /snowflake/session/token, together with SNOWFLAKE_ACCOUNT; then a [default] entry in the platform-rendered $SNOWFLAKE_HOME/connections.toml. Deployments differ in which of those they set, so code that reads only one of them fails where a credential is in fact present.

It also supplies three things hand-rolled connector code usually gets wrong: SNOWFLAKE_ACCOUNT_URL as a bare host, since the connector wants no scheme; role from SNOWFLAKE_ROLE and warehouse from SNOWFLAKE_WAREHOUSE, which are otherwise not applied, leaving the connector on the user’s default role; and a default login_timeout, so a closed-egress sandbox fails fast instead of hanging. You don’t need to set up TLS trust: the sandbox already trusts the platform’s egress proxy.

The helper is for in-sandbox use only, because the credential it reads exists only inside a sandbox. It refuses keyword overrides that would redirect the credential (account, host, token), raising ValueError; non-credential options such as warehouse, database, schema, and role pass through to connect().

When you need to build the connection yourself, see the in-sandbox example in Connect to Snowflake from the SDK, which covers the host handling the helper does for you.

Exceptions¶

Most errors raised by the SDK derive from SandboxError. Two categories fall outside that hierarchy: configuration problems raise RuntimeError (unconfigured client, unresolvable connection), and some transport/timeout conditions surface as RuntimeError (event loop closed) or asyncio.TimeoutError. Guard those separately.

| SandboxContractWarning | The server sent something the SDK does not recognize. Emitted as a warning rather than silently guessing. |

For error handling guidance including what to do for each error, see Cortex Sandboxes error reference.