Bashkit

Sandboxed bash interpreter for Python. Native bindings to the bashkit Rust core for fast, in-process execution with a virtual filesystem.

Homepage: bashkit.sh

Features

• Sandboxed execution in-process, without containers or subprocess orchestration
• Full bash syntax: variables, pipelines, redirects, loops, functions, and arrays
• 164 built-in commands including grep, sed, awk, jq, curl, and find
• Persistent interpreter state across calls, including variables, cwd, and VFS contents
• Direct virtual filesystem APIs, constructor mounts, and live host mounts
• Snapshot and restore support on Bash and BashTool
• AI integrations for LangChain, PydanticAI, and Deep Agents

Installation

pip install bashkit

# Optional integrations
pip install 'bashkit[langchain]'
pip install 'bashkit[pydantic-ai]'
pip install 'bashkit[deepagents]'

Quick Start

Sync Execution

from bashkit import Bash

bash = Bash()

result = bash.execute_sync("echo 'Hello, World!'")
print(result.stdout)  # Hello, World!

bash.execute_sync("export APP_ENV=dev")
print(bash.execute_sync("echo $APP_ENV").stdout)  # dev

Async Execution

import asyncio
from bashkit import Bash


async def main():
    bash = Bash()

    result = await bash.execute("echo -e 'banana\\napple\\ncherry' | sort")
    print(result.stdout)  # apple\nbanana\ncherry

    await bash.execute("printf 'data\\n' > /tmp/file.txt")
    saved = await bash.execute("cat /tmp/file.txt")
    print(saved.stdout)  # data


asyncio.run(main())

Configuration

Constructor Options

from bashkit import Bash

bash = Bash(
    username="agent",
    hostname="sandbox",
    cwd="/home/agent",
    env={"LANG": "en_US.UTF-8"},
    max_commands=1000,
    max_loop_iterations=10000,
    max_memory=10 * 1024 * 1024,
    timeout_seconds=30,
    python=False,
)

Embedded Python and TypeScript

Native wheels include both runtimes, disabled by default. Enable only those your application needs:

from bashkit import BashTool

# Python also requires host opt-in; scripts cannot set this permission.
tool = BashTool(
    python=True,
    typescript=True,
    env={"BASHKIT_ALLOW_INPROCESS_PYTHON": "1"},
)
assert tool.execute_sync("python -c 'print(42)'").stdout == "42\n"
assert tool.execute_sync("ts -c '42'").stdout == "42\n"

LangChain's BashkitTool and deepagents' middleware/backend accept the same flags. Runtime flags survive reset(); supply them explicitly when restoring with BashTool.from_snapshot() or BashTool.from_snapshot_keyed().

Live Output

from bashkit import Bash

bash = Bash()


def on_output(stdout: str, stderr: str) -> None:
    if stdout:
        print(stdout, end="", flush=True)
    if stderr:
        print(stderr, end="", flush=True)


result = bash.execute_sync(
    "for i in 1 2 3; do echo out-$i; echo err-$i >&2; done",
    on_output=on_output,
)

on_output is optional and fires during execution with chunked (stdout, stderr) pairs. Chunks are not line-aligned or exact terminal interleaving, but concatenating all callback chunks matches the final ExecResult.stdout and ExecResult.stderr. The handler must be synchronous; async def callbacks and callbacks that return awaitables are rejected.

Virtual Filesystem

Direct Methods on Bash and BashTool

from bashkit import Bash

bash = Bash()
bash.mkdir("/data", recursive=True)
bash.write_file("/data/config.json", '{"debug": true}\n')
bash.append_file("/data/config.json", '{"trace": false}\n')

print(bash.read_file("/data/config.json"))
print(bash.exists("/data/config.json"))
print(bash.ls("/data"))
print(bash.glob("/data/*.json"))

The same direct filesystem helpers are available on BashTool().

FileSystem Accessor

from bashkit import Bash

bash = Bash()
fs = bash.fs()

fs.mkdir("/data", recursive=True)
fs.write_file("/data/blob.bin", b"\x00\x01hello")
fs.copy("/data/blob.bin", "/data/backup.bin")

assert fs.read_file("/data/blob.bin") == b"\x00\x01hello"
assert fs.exists("/data/backup.bin")

Native Extension Interop

from bashkit import Bash, FileSystem

source = FileSystem()
source.write_file("/org/repo/README.md", b"hello\n")

capsule = source.to_capsule()
imported = FileSystem.from_capsule(capsule)

bash = Bash()
bash.mount("/workspace", imported)
print(bash.execute_sync("cat /workspace/org/repo/README.md").stdout)

to_capsule() / from_capsule() exchange a versioned stable ABI handle so separate PyO3 extensions do not need to share bashkit's internal Rust object layout. Native extensions should depend on bashkit with the interop feature and use bashkit::interop::fs.

Pre-Initialized Files

from bashkit import Bash

bash = Bash(
    files={
        "/config/static.txt": "ready\n",
        "/config/report.json": lambda: '{"ok": true}\n',
    }
)

print(bash.execute_sync("cat /config/static.txt").stdout)
print(bash.execute_sync("cat /config/report.json").stdout)

Real Filesystem Mounts

from bashkit import Bash

bash = Bash(
    mounts=[
        {"host_path": "/path/to/data", "vfs_path": "/data"},
        {"host_path": "/path/to/workspace", "vfs_path": "/workspace", "writable": True},
    ]
)

print(bash.execute_sync("ls /workspace").stdout)

Live Mounts

from bashkit import Bash, FileSystem

bash = Bash()
workspace = FileSystem.real("/path/to/workspace", writable=True)

bash.mount("/workspace", workspace)
bash.execute_sync("echo 'hello' > /workspace/demo.txt")
bash.unmount("/workspace")

Network Access

curl, wget, and http are gated behind an explicit allowlist. Without a network= kwarg outbound HTTP is disabled (the secure default). Pass an explicit allowlist or allow_all=True to opt in:

from bashkit import Bash

# Per-host allowlist - all other URLs are blocked.
bash = Bash(network={"allow": ["https://api.github.com", "https://api.openai.com/v1"]})

# Allow every URL (mirrors NetworkAllowlist::allow_all() in the Rust core).
trusted = Bash(network={"allow_all": True})

# Disable the SSRF guard if you legitimately need to reach a private IP.
local = Bash(network={"allow": ["http://127.0.0.1:8080"], "block_private_ips": False})

network= is also accepted by BashTool(...) and persists across reset() and from_snapshot(...). An explicit network={"allow": []} blocks every URL but is distinct from omitting network= entirely.

Credential injection

Inject per-host credentials transparently so scripts never see the real secret. Two modes are supported (mirroring the Rust BashBuilder::credential and BashBuilder::credential_placeholder APIs):

from bashkit import Bash

bash = Bash(
    network={
        "allow": ["https://api.github.com", "https://api.openai.com/v1"],
        # Direct injection — script has no knowledge of the credential.
        "credentials": [
            {
                "pattern": "https://api.github.com",
                "kind": "bearer",
                "token": "ghp_xxx",
            },
        ],
        # Placeholder mode — script sees an opaque placeholder via env var.
        "credential_placeholders": [
            {
                "env": "OPENAI_API_KEY",
                "pattern": "https://api.openai.com",
                "kind": "bearer",
                "token": "sk-real-key",
            },
        ],
    },
)
# Scripts can: curl -s https://api.github.com/repos/foo/bar
#   → Authorization: Bearer ghp_xxx is added on the wire.
# Scripts can: curl -s -H "Authorization: Bearer $OPENAI_API_KEY" \
#                   https://api.openai.com/v1/chat
#   → the placeholder is replaced with sk-real-key on the wire.

Each credential dict accepts kind: "bearer" (token), kind: "header" (name, value), or kind: "headers" (a list of (name, value) pairs). Injected headers overwrite script-provided headers with the same name to prevent credential spoofing. Phase 2 of #1348 covers the credential surface; request callbacks and bot-auth ship in follow-ups.

Error Handling

from bashkit import Bash, BashError

bash = Bash()

try:
    bash.execute_sync_or_throw("exit 42")
except BashError as err:
    print(err.exit_code)  # 42
    print(err.stderr)
    print(str(err))

Use execute_or_throw() and execute_sync_or_throw() when you want failures surfaced as exceptions instead of inspecting exit_code manually.

Cancellation

from bashkit import Bash

bash = Bash()

bash.cancel()  # abort in-flight execution (no-op if idle)
bash.clear_cancel()  # clear the sticky flag so subsequent executions work

cancel() sets a sticky flag that causes every future execute() to fail immediately with "execution cancelled". Call clear_cancel() after the cancelled execution finishes to restore the instance for reuse, this preserves all VFS state. Use reset() only when you want to discard VFS and shell state entirely.

BashTool exposes the same cancel(), clear_cancel(), and reset() methods.

Shell State

from bashkit import Bash

bash = Bash()
bash.execute_sync("mkdir -p /workspace && cd /workspace")

state = bash.shell_state()
prompt = f"{state.cwd}$ "
print(prompt)  # /workspace$

bash.execute_sync("cd /")
print(state.cwd)  # still /workspace

ShellState is a read-only snapshot for prompt rendering and inspection. It is a Python-friendly inspection view rather than a full Rust-shell mirror, and fields like env, variables, and arrays are exposed as immutable mappings. Use snapshot(exclude_filesystem=True) when you need shell-only restore bytes, or snapshot(exclude_filesystem=True, exclude_functions=True) when prompt rendering does not need function restore. Transient fields like last_exit_code and traps are captured on the snapshot, but the next top-level execute() / execute_sync() clears them before running the new command. BashTool exposes the same shell_state() method.

Script Analysis

analyze() reports what a script statically refers to, commands, arguments, redirect targets, function definitions, without running it. Use it to decide whether a model-produced command needs user approval.

from bashkit import Bash, BashError

bash = Bash()
analysis = bash.analyze("cat notes.txt | grep -i todo > out.txt")

analysis.command_names  # ["cat", "grep"]
analysis.commands[1].args  # ["-i", "todo"]
analysis.redirects[0].path  # "out.txt"
analysis.redirects[0].is_write  # True
analysis.is_opaque  # False

Words that are not fully literal report None, never a partial string:

bash.analyze('rm "$target" /tmp/fixed').commands[0].args  # [None, "/tmp/fixed"]

Advisory only. Static analysis cannot see through dynamic dispatch, eval, functions, or aliases. Those set is_opaque, and an allowlist check must consult it, c=rm; $c -rf /data contains no disallowed command name:

READ_ONLY = {"ls", "cat", "head", "grep", "wc", "echo"}


def safe_to_run(bash, script):
    try:
        analysis = bash.analyze(script)  # raises BashError if it does not parse
    except BashError:
        return False
    if analysis.is_opaque:
        return False
    return all(c.is_assignment_only or (c.name is not None and c.name in READ_ONLY) for c in analysis.commands)

Commands inside $(…), loops, branches, and function bodies are reported too, each tagged with context ("direct", "substitution", "function_body"), so echo $(rm -rf /) never looks like a bare echo. Enforcement stays with the builtin registry, the network allowlist, and the mount policy.

BashTool

BashTool wraps Bash and adds tool-contract metadata for agent frameworks:

• name
• short_description
• version
• description()
• help()
• system_prompt()
• input_schema()
• output_schema()

from bashkit import BashTool

tool = BashTool()

print(tool.description())
print(tool.input_schema())

result = tool.execute_sync("echo 'Hello from BashTool'")
print(result.stdout)

Python Function Calls

Register host functions for code running in embedded Python. Both Bash and BashTool accept external_functions and an async external_handler:

from bashkit import BashTool


async def handler(name, args, kwargs):
    if name == "lookup":
        return {"value": args[0] * kwargs["multiplier"]}
    raise ValueError(f"Unknown function: {name}")


tool = BashTool(
    python=True,
    env={"BASHKIT_ALLOW_INPROCESS_PYTHON": "1"},
    external_functions=["lookup"],
    external_handler=handler,
)
result = await tool.execute("python -c 'print(lookup(21, multiplier=2)[\"value\"])'")
assert result.stdout == "42\n"

LangChain's create_bash_tool / BashkitTool and deepagents' middleware/backend accept the same options. ainvoke() and BashkitBackend.aexecute() run callbacks on the caller's active asyncio loop, preserving loop-bound clients and request ContextVars. Use these async APIs in FastAPI async endpoints or notebook cells with top-level await.

Synchronous execution works from ordinary threads, FastAPI sync endpoints, and notebook cells with an already-running loop. It drives callbacks on a dedicated private worker loop shared across synchronous calls to that instance. Clients bound to another loop must use the async API. ContextVars are captured for each execution; a shared instance serializes requests. Timeouts and cancellation cancel that execution's pending callback tasks.

If another handler is already waiting on the current loop, synchronous access to that same interpreter fails immediately: blocking the loop would deadlock the handler. Use the async API or move the synchronous operation to asyncio.to_thread().

Callbacks are trusted host code: enforce service authorization and I/O limits there. They cannot re-enter their own interpreter or live filesystem handles; pass data through arguments/results or use another instance. Function names and handlers survive reset() and in-place restore. Snapshot bytes do not serialize callbacks: supply them again to from_snapshot() / from_snapshot_keyed().

Runnable notebook: runtime_external_functions.ipynb.

TypeScript Function Calls

BashTool and its LangChain/deepagents adapters also accept separate typescript_external_functions / typescript_external_handler options:

async def ts_handler(name, args):
    return {"value": args[0] * args[1]["multiplier"]}


tool = BashTool(
    typescript=True,
    typescript_external_functions=["lookup"],
    typescript_external_handler=ts_handler,
)
result = await tool.execute("ts -c 'const r = await lookup(21, {multiplier: 2}); r.value'")
assert result.stdout == "42\n"

TypeScript dispatch receives (name, args); use an object argument for named options. Values cross as None/null, booleans, signed 64-bit integers, finite floats, strings, lists, and string-keyed dictionaries. undefined arguments map to None; function arguments, unsupported return objects, non-finite numbers, and nesting deeper than 64 levels are rejected. Python and TypeScript handlers can coexist, share the VFS, and use the same thread/loop/reset/snapshot rules. Native host handlers are unavailable in the Pyodide build.

ZapCode currently loses console.log() output produced after an external call. Return the final expression as shown above, or write results into the VFS.

Persistent Custom Builtins

Use custom_builtins={...} on Bash or BashTool when you want constructor-time Python callback builtins without giving up persistent shell/VFS state:

from bashkit import Bash
import json


def get_order(ctx):
    if not ctx.argv or ctx.argv[0] in {"help", "--help"}:
        return "usage: get-order <get|help> [args]\n"
    if ctx.argv[0] == "get" and len(ctx.argv) >= 2:
        return json.dumps({"id": ctx.argv[1], "status": "shipped", "items": ["widget"]}) + "\n"
    return "usage: get-order <get|help> [args]\n"


bash = Bash(custom_builtins={"get-order": get_order})

bash.execute_sync("mkdir -p /scratch && get-order get 42 > /scratch/order.json")
result = bash.execute_sync("cat /scratch/order.json | jq -r '.items[]'")
print(result.stdout)  # widget

Callbacks receive a BuiltinContext object with raw argv tokens, optional pipeline stdin, the current cwd, and visible env. They may return either the builtin stdout string directly or a BuiltinResult(stdout=..., stderr=..., exit_code=...) for explicit shell-shaped failures. Raise an exception for unexpected callback failures. Async callbacks support the same return shapes. BashTool exposes the same custom_builtins constructor kwarg and includes registered command names in help() output for LLM-facing metadata.

from bashkit import BuiltinResult


def view_image(ctx):
    if not ctx.argv:
        return BuiltinResult(stderr="view-image: missing path\n", exit_code=1)
    return BuiltinResult(stdout="")

When you use await bash.execute(...) or await bash_tool.execute(...), async callbacks are scheduled back onto the caller's active asyncio loop, so loop-bound primitives like asyncio.Event and framework-owned clients keep working. execute_sync() still supports async callbacks, but it drives them on an internal private loop and should not be called from an async endpoint.

ScriptedTool

Use ScriptedTool to register Python callbacks as bash-callable tools:

from bashkit.scripted import ScriptedTool


def get_user(params, stdin=None):
    return '{"id": 1, "name": "Alice"}'


tool = ScriptedTool("api")
tool.add_tool(
    "get_user",
    "Fetch user by ID",
    callback=get_user,
    schema={"type": "object", "properties": {"id": {"type": "integer"}}},
)

result = tool.execute_sync("get_user --id 1 | jq -r '.name'")
print(result.stdout)  # Alice

ScriptedTool callbacks receive (params_dict, stdin_or_none) and must return the tool stdout string. Async callbacks are also supported. await tool.execute(...) runs async callbacks on the caller's active asyncio loop; execute_sync() falls back to a private loop because there is no caller loop to reuse.

Snapshot / Restore

from bashkit import Bash

bash = Bash(username="agent", max_commands=100)
bash.execute_sync(
    'export BUILD_ID=42; greet() { echo "hi $1"; }; mkdir -p /workspace && cd /workspace && echo ready > state.txt'
)

snapshot = bash.snapshot()
shell_only = bash.snapshot(exclude_filesystem=True)
prompt_only = bash.snapshot(exclude_filesystem=True, exclude_functions=True)
trusted_snapshot = bash.snapshot_keyed(b"32+ bytes of application secret")

restored = Bash.from_snapshot(snapshot, username="agent", max_commands=100)
assert restored.execute_sync("echo $BUILD_ID").stdout.strip() == "42"
assert restored.execute_sync("greet agent").stdout.strip() == "hi agent"
assert restored.execute_sync("cat /workspace/state.txt").stdout.strip() == "ready"

restored.reset()
restored.restore_snapshot(snapshot)
assert restored.execute_sync("pwd").stdout.strip() == "/workspace"
restored.restore_snapshot(shell_only)
restored.restore_snapshot_keyed(trusted_snapshot, b"32+ bytes of application secret")

Unkeyed snapshot() bytes detect accidental corruption only. Use snapshot_keyed(...), restore_snapshot_keyed(...), and from_snapshot_keyed(...) whenever snapshot bytes cross trust boundaries such as uploads, shared storage, or network transport.

BashTool exposes the same snapshot, restore, and keyed snapshot APIs. Python callback builtins are host-side config, not serialized shell state, so pass custom_builtins= again when constructing a restored instance if you need them after snapshot restore.

Framework Integrations

LangChain

from bashkit.langchain import create_bash_tool

tool = create_bash_tool(
    mounts=[
        {
            "host_path": "/path/to/docs",
            "vfs_path": "/docs",
            "writable": False,
        }
    ],
    allowed_mount_paths=["/path/to/docs"],
    readonly_filesystem=True,
    max_output_length=8_000,
)

PydanticAI

from bashkit.pydantic_ai import create_bash_tool

tool = create_bash_tool()

Deep Agents

Requires Python 3.11+ and pip install 'bashkit[deepagents]'.

from bashkit.deepagents import BashkitBackend

backend = BashkitBackend(timeout_seconds=30)
backend.write("/tmp/example.py", "print(42)")
assert backend.read("/tmp/example.py").file_data["content"] == "print(42)"
assert backend.grep("print", "/tmp", glob="*.py").matches[0]["line"] == 1
middleware = backend.create_middleware()  # Shares the same virtual filesystem

Uses the structured Deep Agents 0.7.13+ backend protocol. Shell execution reports native output truncation; file operations preserve exact VFS contents. Configure execution timeouts on the backend constructor.

API Reference

Bash

• execute(commands: str) -> ExecResult
• execute_sync(commands: str) -> ExecResult
• execute_or_throw(commands: str) -> ExecResult
• execute_sync_or_throw(commands: str) -> ExecResult
• cancel()
• clear_cancel()
• reset()
• constructor kwarg: custom_builtins={name: callback}
• snapshot() -> bytes
• snapshot_keyed(key: bytes) -> bytes
• restore_snapshot(data: bytes)
• restore_snapshot_keyed(data: bytes, key: bytes)
• from_snapshot(data: bytes, **kwargs) -> Bash
• from_snapshot_keyed(data: bytes, key: bytes, **kwargs) -> Bash
• constructor kwarg: network={"allow": [...], "block_private_ips": True} or network={"allow_all": True}, optionally with credentials=[...] and credential_placeholders=[...]
• mount(vfs_path: str, fs: FileSystem)
• unmount(vfs_path: str)
• Direct VFS helpers: read_file, write_file, append_file, mkdir, remove, exists, stat, read_dir, ls, glob, copy, rename, symlink, chmod, read_link
• analyze(script: str) -> ScriptAnalysis, static, pre-execution introspection (see Script Analysis)

BashTool

• All execution, cancellation (cancel(), clear_cancel()), reset, snapshot, restore, mount, and direct VFS helpers from Bash
• constructor kwarg: custom_builtins={name: callback}
• Tool metadata: name, short_description, version
• description() -> str
• help() -> str
• system_prompt() -> str
• input_schema() -> str
• output_schema() -> str

ScriptAnalysis

Returned by analyze(script).

• commands: list[AnalyzedCommand], every simple command, in source order
• redirects: list[AnalyzedRedirect], path: str | None, mode: "read" | "write" | "append", is_write: bool
• functions: list[str], function names the script defines
• command_names: list[str], distinct statically known names, first-seen order
• has_dynamic_commands / has_interpreter_reentry / has_command_substitution / truncated
• is_opaque: bool, the script hides work; allowlist checks must treat this as "ask"
• commands_named(name) -> list[AnalyzedCommand], to_dict() -> dict

AnalyzedCommand: name: str | None, args: list[str | None], context: str, assignments: list[str], is_assignment_only: bool

ScriptedTool

• add_tool(name, description, callback, schema=None)
• execute(script: str) -> ExecResult
• execute_sync(script: str) -> ExecResult
• execute_or_throw(script: str) -> ExecResult
• execute_sync_or_throw(script: str) -> ExecResult
• env(key: str, value: str)
• tool_count() -> int

FileSystem

• mkdir(path, recursive=False)
• write_file(path, content)
• read_file(path) -> bytes
• append_file(path, content)
• exists(path) -> bool
• remove(path, recursive=False)
• stat(path) -> dict
• read_dir(path) -> list
• rename(src, dst)
• copy(src, dst)
• symlink(target, link)
• chmod(path, mode)
• read_link(path) -> str
• FileSystem.real(host_path, writable=False) -> FileSystem
• FileSystem.from_capsule(capsule) -> FileSystem
• FileSystem.to_capsule() -> object

ExecResult and BashError

• ExecResult.stdout
• ExecResult.stderr
• ExecResult.exit_code
• ExecResult.error
• ExecResult.success
• ExecResult.to_dict()
• BashError.exit_code
• BashError.stderr

Platform Support

• Linux: x86_64, aarch64 (glibc and musl wheels)
• macOS: x86_64, aarch64
• Windows: x86_64
• Python: 3.9 through 3.14

How It Works

Bashkit is built on the bashkit Rust core, which implements a sandboxed bash interpreter and virtual filesystem. The Python package exposes that engine through a native extension, so commands run in-process with persistent state and resource limits, without shelling out to the host system.

Part of Everruns

Bashkit is part of the Everruns ecosystem. See the bashkit monorepo for the Rust core, the JavaScript package (@everruns/bashkit), and related tooling.

License

MIT