Parselbox

import asyncioimport osfrom textwrap import dedentfrom parselbox import Parselboxfrom parselbox.bridge import HTTPBridge, ShellBridge
class Analytics:    def summarize(self, repos: list) -> dict:        """Aggregate repo stats."""        stars = [r["stars"] for r in repos]        return {"count": len(repos), "avg_stars": round(sum(stars) / len(stars))}
config = {"mcpServers": {"playwright": {"command": "npx", "args": ["@playwright/mcp@latest"]}}}
async def main():    async with Parselbox(        mcp=config,        context={            "analytics": Analytics(),            "github": HTTPBridge(base_url="https://api.github.com", token=os.environ["GITHUB_TOKEN"]),            "sh": ShellBridge("bash"),        },        network=True,        allow_runtime_packages=True,        packages=["numpy", "npm:lodash"],        output_dir="./workspace",    ) as sbx:        # Discover available tools        await sbx.execute_code("sbx.search('navigate|get')")
        # Scrape Hacker News for GitHub links in a real browser        await sbx.execute_code(dedent("""            import re            playwright.browser_navigate(url="https://news.ycombinator.com")            text = playwright.browser_snapshot()            repos = re.findall(r'github\\.com/([\\w.-]+/[\\w.-]+)', text)[:5]        """))
        # Fetch star counts in parallel, then summarize via the context bridge        await sbx.execute_code(dedent("""            import asyncio            results = await asyncio.gather(*[github.get.task(f"/repos/{r}") for r in repos])            repo_data = [{"name": r["data"]["name"], "stars": r["data"]["stargazers_count"]}                         for r in results if r.get("ok")]            analytics.summarize(repo_data)        """))
        # Chart it — matplotlib auto-installs on import        result = await sbx.execute_code(dedent("""            import matplotlib.pyplot as plt            plt.barh([r["name"] for r in repo_data], [r["stars"] for r in repo_data])            plt.savefig("chart.png")        """))        print(result.files)                  # ['chart.png']        image = sbx.read_file("chart.png")        # every result carries .output, .files, .stdout, .stderr, .error
        # Serve the whole sandbox as an MCP server        await sbx.run_mcp()
asyncio.run(main())

import asynciofrom parselbox import Parselboxfrom agents import Agent, Runner, function_tool
sandbox = Parselbox(    mcp={"mcpServers": {"playwright": {"command": "npx", "args": ["@playwright/mcp@latest"]}}},    output_dir="./outputs",    allow_runtime_packages=True,)
agent = Agent(    name="Research Assistant",    model="gpt-5.5",    instructions=f"You are a world-class research assistant.\n\n{sandbox.get_prompt()}",    tools=[function_tool(sandbox.get_tool())],)
async def main():    async with sandbox:        result = await Runner.run(            agent,            "Scrape Wikipedia's 'List of highest-grossing films' with the Playwright MCP. "            "Plot a bar chart of the top 10 and save it as ./plot.png",            max_turns=30,        )        print(result.final_output)
asyncio.run(main())

from parselbox import Parselbox
class Calculator:    def add(self, a: float, b: float) -> float:        """Add two numbers."""        return a + b
async with Parselbox(context={"calc": Calculator()}) as sbx:    await sbx.execute_code("calc.add(a=10, b=20)")

from parselbox import Parselboxfrom parselbox.bridge import Bridgefrom pydantic import BaseModel
class Coordinate(BaseModel):    x: float    y: float    z: float = 0.0
class Sensors(Bridge):    def temperature(self) -> float:        """Read temperature in celsius."""        return 23.5
class Robot(Bridge):    def __init__(self):        self.sensors = Sensors()
    def move(self, to: Coordinate) -> dict:        """Move robot to a position."""        return {"position": [to.x, to.y, to.z], "status": "reached"}
async with Parselbox(context={"robot": Robot()}) as sbx:    await sbx.execute_code("robot.move(to={'x': 1, 'y': 2})")    await sbx.execute_code("robot.sensors.temperature()")

from parselbox import Parselboxfrom parselbox.bridge import HTTPBridge, GraphQLBridge, ShellBridge
api = HTTPBridge(    spec="https://petstore3.swagger.io/api/v3/openapi.json",    base_url="https://petstore3.swagger.io/api/v3",)gql = GraphQLBridge("https://countries.trevorblades.com/graphql")sh = ShellBridge("ssh -T user@host")
mcp = {"mcpServers": {"deepwiki": {"type": "http", "url": "https://mcp.deepwiki.com/mcp"}}}
async with Parselbox(context={"api": api, "gql": gql, "sh": sh}, mcp=mcp, network=True) as sbx:    await sbx.execute_code('api.search("GET /pet/*")')    await sbx.execute_code('api.get("/pet/1")')
    await sbx.execute_code('gql.graphql(query="{ continents { name } }")')    await sbx.execute_code('gql.graphql(query="{ languages { code name } }")')
    await sbx.execute_code('term = sh.shell.task()')    await sbx.execute_code('term.send("df -h")')
    await sbx.execute_code("sbx.search('ask|read')")    await sbx.execute_code("deepwiki.read_wiki_structure(repoName='pyodide/pyodide')")    await sbx.execute_code("deepwiki.ask_question(question='What is Pyodide?', repoName='pyodide/pyodide')")

job = sh.exec.task(command="ffmpeg -i in.mp4 out.mp4")   # returns a task immediately
job.status()                   # TaskStatus(state, elapsed, message, logfile)job.tail(5)                    # last lines of the task's live logjob.send("q")                  # message a running interactive processawait job.wait(timeout=120)    # block until done — or just `await job`job.cancel()
# parallel fan-outimport asyncioresults = await asyncio.gather(*[api.get.task(f"/items/{i}") for i in range(5)])

from parselbox.bridge import Bridge
class Exporter(Bridge):    def run(self, rows: int) -> str:        for i in range(rows):            self.log(f"row {i}/{rows}")     # appended to task.logfile → tail()            for msg in self.recv():         # messages queued by task.send()                self.log(f"got: {msg}")        return "done"

session = sh.shell.task()               # a live shell — state persists within the sessionsession.send("x=21")session.send("echo $((x * 2))")
import asyncioawait asyncio.sleep(1)                  # give it a beatsession.tail(1)                         # "42"
session.cancel()

from parselbox import Parselbox, Mount
async with Parselbox(    files=["data.csv"],                         # Read/write at /files/data.csv    mounts=[        Mount("./datasets", "/data", "ro"),     # Read-only at /mnt/data        Mount("./workspace", "/work", "rw"),    # Read/write at /mnt/work    ],    output_dir="./outputs"                      # Sandbox files persisted here) as sandbox:    # Write a file into the sandbox from the host    sandbox.write_file("greeting.txt", "Hello from host!")
    code = """    content = open('/files/data.csv').read()               # input file    ref = open('/mnt/data/reference.json').read()          # read-only mount    open('/mnt/work/processed.txt', 'w').write(content)    # read/write mount    open('result.txt', 'w').write("Done!")                 # working dir -> output_dir    """    result = await sandbox.execute_code(code)
    # New / modified files are detected and returned    print(result.files)   # ['result.txt', 'greeting.txt']    sandbox.read_file("result.txt")

bash("echo 'hello from bash' > note.txt && cat note.txt")   # shell over the workspace

from parselbox import Parselbox, Mount
# preload Python + npm packages on startupParselbox(packages=["numpy", "pandas", "npm:lodash"])
# local wheel — mount its dir so Deno can read the host pathParselbox(packages=["file:///host/wheels/pkg.whl"],          mounts=[Mount("./wheels", "wheels", "ro")])
# remote wheel — needs network accessParselbox(packages=["https://example.com/pkg.whl"], network=True)
# autoload as imports appear (only official domains when network=False)Parselbox(allow_runtime_packages=True)

# Block everything (default)Parselbox(network=False)
# Allow specific domains (Python API only)Parselbox(network=["api.github.com:443"])
# Allow everythingParselbox(network=True)

async with Parselbox(    network=["127.0.0.1:8900"],                 # sandbox can ONLY reach the proxy    env={        "OPENAI_BASE_URL": "http://127.0.0.1:8900/v1",        "OPENAI_API_KEY": "phantom-token",      # harmless; the real key lives on the proxy    },) as sbx:    await sbx.execute_code("import openai; openai.OpenAI().chat.completions.create(...)")

# Basic — auto converts args and resultsjs("return data.map(x => x * 2)", data=[1, 2, 3])  # [2, 4, 6]
# Callbacks — Python functions auto-proxied, no create_proxy neededjs("return items.filter(fn)", items=[1,2,3,4,5], fn=lambda x, *_: x > 3)  # [4, 5]
# Async + Web APIs (Intl, Crypto, URL, TextEncoder)js("return crypto.randomUUID()")

# npm packages — returns proxy + auto-injects in js() scope (alias= to rename)lodash = require("lodash")lodash.chunk([1, 2, 3, 4], 2)  # [[1, 2], [3, 4]]
# Callbacks work with require'd packageslodash.sortBy(data, lambda x, *_: x["age"])
# Also available in js()js("return lodash.invert({a: 1, b: 2})")
# Local TypeScript — compiled by Deno, hot-reloads; can import npm internallyrequire("./math_utils.ts").fibonacci(10)
# .wasm modules & WASI binaries load too — see WASM Tools
# Instances keep their methods — chain themdayjs = require("dayjs")dayjs("2026-06-15").add(30, "day").format("YYYY-MM-DD")   # "2026-07-15"
# Class constructors auto-detect `new`color = require("color")color("red").darken(0.5).hex()   # "#800000"
# Chains work with Python callbackslodash(data).filter(lambda x, *_: x["pay"] > 100).sortBy(lambda x, *_: -x["pay"]).value()

js("""    const path = resolvePath("sample.txt");    const info = await Deno.stat(path);    return { size: info.size, isFile: info.isFile };""")

# Python module — write it, import itopen("helpers.py", "w").write("def double(x): return x * 2")from helpers import doubledouble(21)  # 42
# TypeScript module — compiled by Denoopen("transform.ts", "w").write("export function upper(s: string) { return s.toUpperCase(); }")require("./transform.ts").upper("hello")  # "HELLO"

bash("echo hello > note.txt && cat note.txt | tr a-z A-Z")   # "HELLO"bash("grep -rn hello . | wc -l")bash("curl -s https://api.github.com/zen")                   # network rules still apply

# Library module (no imports) — its exports become methodsrequire("./fib.wasm").fib(20)                      # 6765
# Command module (a WASI program) — becomes a callable commandpandoc = require("./pandoc.wasm")r = pandoc(["-f", "markdown", "-t", "html5"], stdin="# Report")r["stdout"].decode()                               # '<h1 id="report">Report</h1>'

run(args=None, stdin="", env=None, preopens=None, argv0=None)

open("bin/pandoc.wasm", "wb").write(pandoc_bytes)
bash("pandoc -f markdown -t plain notes.md | head -3 | tr a-z A-Z")#     ^^ compiled pandoc                      ^^ just-bash builtins

# Sandbox guidehelp()
# Namespace tree view — shows all methods with hierarchyhelp(robot)# Remote namespace 'robot' — methods execute on the host and return results.# Methods:# ├── sensors# │   └── temperature()# └── move()
# Tool details — description, parameters, output schemahelp(robot.move)# {"description": "Move robot to position.", "parameters": {...}, "output": {...}}
# Works on local objects toohelp(len)

# Discover what's availablesbx.info()
# Search for tools across all namespacessbx.search("repo|query")
# Get tool signatures before callingsbx.inspect(["github.search_repositories", "db.query", "robot.move"])
# Parallel execution with .taskimport asyncioresults = await asyncio.gather(*[api.fetch.task(id=i) for i in ids])
# Inspect unknown response structuresbx.preview(results)

await sbx.execute_code("""    display("<h1>Q3 Revenue</h1><p class='text-lg'>Up <b>12%</b> to $4.1M</p>")""")

result = await sbx.execute_code('display("<b>done</b>")')result.view          # full HTML document, or None if display() wasn't called

# SDKsandbox = Parselbox(serve=3000)
# CLIuvx parselbox --serve 3000

open("index.html", "w").write("<h1>Hello World</h1>")open("style.css", "w").write("h1 { color: blue; }")

@api.get("/items")def list_items(params):    limit = int(params.get("limit", 10))    return items[:limit]
@api.post("/items")def create_item(body):    return {"id": len(items) + 1, "name": body["name"]}

@api.get("/dashboard")async def dashboard(params):    import asyncio    sensors, orders = await asyncio.gather(        robot.sensors.temperature.task(),        store.get.task("/orders", params={"limit": 5}),    )    return {"temperature": sensors, "recent_orders": orders}

from parselbox import Parselbox, Callback, ExecutionResultfrom parselbox.hooks import Hook
class AuditHook(Hook):    async def pre_execute(self, code: str):        print(f"Executing: {code[:80]}...")
    async def post_execute(self, result: ExecutionResult):        print(f"Result: {result.output}")
    async def pre_tool_call(self, callback: Callback):        if "drop" in str(callback.kwargs).lower():            raise PermissionError("DROP statements are blocked")
    async def post_tool_call(self, callback: Callback, result):        print(f"Tool {callback.name} returned")
async with Parselbox(    context={"db": db},    hooks=[AuditHook()],) as sbx:    await sbx.execute_code("db.query(sql='SELECT 1')")

from parselbox import Parselbox, Mount
sandbox = Parselbox(    context=dict(db=db, notify=send_alert),   # Proxied functions and namespaces    globals=dict(name="hi", threshold=0.5),   # Static values copied into sandbox    files=["./input.txt"],                    # Read/write files at /files/    mounts=[        Mount("./datasets", "/data", "ro"),   # Read-only mount        Mount("./workspace", "/work", "rw"),  # Read/write mount    ],    output_dir="./outputs",                   # Persist sandbox files    packages=["numpy", "npm:lodash"],         # Install on startup (Python + npm)    package_dir="./cache",                    # Persist package cache across sessions    allow_runtime_packages=True,              # Auto-install from imports (default: False)    network=True,                             # True, False, or ["domain:port", ...] (API only)    mcp="./mcp.json",                         # Connect MCP servers (path or dict)    serve=8080,                               # Enable web server on port    memory=2048,                              # WASM memory limit in MB (default: 2048)    timeout=60,                               # Execution timeout in seconds (default: 60, 0 disables)    hooks=[AuditHook()],                      # Lifecycle hooks    env={                                     # Custom env vars (available in Python os.environ)        "OPENAI_BASE_URL": "http://proxy/v1", # SDK base_url overrides for reverse proxy        "OPENAI_API_KEY": "phantom",          # Phantom tokens (real keys on proxy)        "HTTP_PROXY": "http://proxy:8080",    # Deno-level proxy (filtered from os.environ)        "DENO_CERT": "/path/to/ca.pem",       # Custom CA for MITM proxy    },)