Try Live
Add Docs
Rankings
Pricing
Enterprise
Docs
Install
Install
Docs
Pricing
Enterprise
More...
More...
Try Live
Rankings
Add Docs
AIO Sandbox
https://github.com/agent-infra/sandbox
Admin
AIO Sandbox is an all-in-one agent sandbox environment that unifies browser, shell, file, MCP
...
Tokens:
91,336
Snippets:
884
Trust Score:
5.9
Update:
3 days ago
Context
Skills
Chat
Benchmark
66
Suggestions
Latest
Show doc for...
Code
Info
Show Results
Context Summary (auto-generated)
Raw
Copy
Link
# AIO Sandbox AIO Sandbox is an all-in-one agent sandbox environment that combines Browser automation, Shell terminal, File operations, MCP (Model Context Protocol) services, Jupyter notebooks, and VSCode Server in a single Docker container. Built on cloud-native lightweight sandbox technology, it provides a unified, secure execution environment where all components share the same filesystem — files downloaded in the browser are instantly accessible in Shell or File operations. The project targets AI agents and developers who need a fully pre-configured, reproducible environment without manual setup. The sandbox exposes its functionality through three official SDKs (Python, TypeScript/JavaScript, and Go), a REST API (all endpoints under `/v1/`), and a pre-configured MCP Hub at `/mcp`. The Python package is `agent-sandbox`, the Node.js package is `@agent-infra/sandbox`, both generated from the same OpenAPI definition via Fern. Every major capability — shell execution, file I/O, browser control, Python/JavaScript code execution, tab management, cookie handling, network monitoring, and cloud provider lifecycle management — is accessible through these SDKs. --- ## Installation and Setup Install the sandbox as a Docker container; access all services on port 8080. ```bash # Start the sandbox container docker run --security-opt seccomp=unconfined --rm -it -p 8080:8080 \ ghcr.io/agent-infra/sandbox:latest # Install Python SDK pip install agent-sandbox # Install TypeScript/JavaScript SDK npm install @agent-infra/sandbox # Install Go SDK go get github.com/agent-infra/sandbox-sdk-go ``` Once running, access points: - REST API docs: `http://localhost:8080/v1/docs` - VNC Browser: `http://localhost:8080/vnc/index.html?autoconnect=true` - VSCode Server: `http://localhost:8080/code-server/` - MCP Hub: `http://localhost:8080/mcp` --- ## Client Initialization Initialize either the Python or TypeScript client pointing to the running sandbox. **Python** — `Sandbox(base_url, headers?, timeout?)` connects to the sandbox REST API and exposes all service namespaces as attributes. ```python from agent_sandbox import Sandbox # Basic initialization client = Sandbox(base_url="http://localhost:8080") # With authentication header and custom timeout client = Sandbox( base_url="http://localhost:8080", headers={"Authorization": "Bearer <jwt>"}, timeout=120, ) # Retrieve home directory from sandbox context home_dir = client.sandbox.get_context().home_dir print(home_dir) # e.g. /home/gem ``` **TypeScript** — `SandboxClient({ environment, timeoutInSeconds?, headers? })` with identical service namespaces. ```typescript import { SandboxClient } from '@agent-infra/sandbox'; const client = new SandboxClient({ environment: process.env.SANDBOX_API_URL || 'http://localhost:8080', timeoutInSeconds: 60, }); const ctx = await client.sandbox.getContext(); if (ctx.ok) console.log(ctx.body); // { home_dir: '/home/gem', ... } ``` --- ## sandbox.get_context / sandbox.getContext Returns environment metadata for the running sandbox: home directory, system version, and platform info. ```python from agent_sandbox import Sandbox client = Sandbox(base_url="http://localhost:8080") context = client.sandbox.get_context() print(context.home_dir) # /home/gem print(context.version) # v1.0.0.150 packages = client.sandbox.get_python_packages() print(packages.data) # list of installed Python packages ``` --- ## shell.exec_command / shell.execCommand Execute a shell command synchronously and return stdout/stderr. Supports pipes, environment variables, working directory, and timeouts. ```python from agent_sandbox import Sandbox client = Sandbox(base_url="http://localhost:8080") # Simple command result = client.shell.exec_command(command="echo 'Hello World'") print(result.data.output) # Hello World # Piped command result = client.shell.exec_command( command="ls -la /tmp | head -5" ) print(result.data.output) # Command in a specific directory result = client.shell.exec_command( command="pwd", exec_dir="/home/gem" ) print(result.data.output) # /home/gem ``` ```typescript import { SandboxClient } from '@agent-infra/sandbox'; const client = new SandboxClient({ environment: 'http://localhost:8080' }); const res = await client.shell.execCommand({ command: 'echo "Hello from sandbox!"' }); if (res.ok) console.log(res.body.data.output); // Hello from sandbox! // Piped command const piped = await client.shell.execCommand({ command: 'echo "Hello World" | tr "a-z" "A-Z"', }); if (piped.ok) console.log(piped.body.data.output); // HELLO WORLD ``` --- ## shell.create_session / shell.createSession Create a persistent interactive shell session identified by a session ID. Useful for multi-step workflows that require shared state between commands. ```python import uuid from agent_sandbox import Sandbox client = Sandbox(base_url="http://localhost:8080") # Create a persistent session session = client.shell.create_session( id=str(uuid.uuid4()), exec_dir="/home/gem", no_change_timeout=120, ) session_id = session.data.session_id # Run commands in the same session (state is preserved) client.shell.exec_command(command="export MY_VAR=hello", id=session_id) result = client.shell.exec_command(command="echo $MY_VAR", id=session_id) print(result.data.output) # hello # List active sessions sessions = client.shell.list_sessions() print(sessions.data) # Cleanup when done client.shell.cleanup_session(session_id) ``` --- ## file.write_file / file.writeFile Write text or binary (base64) content to a file path inside the sandbox. Returns the number of bytes written. ```python import base64 from agent_sandbox import Sandbox client = Sandbox(base_url="http://localhost:8080") home_dir = client.sandbox.get_context().home_dir # Write a text file result = client.file.write_file( file=f"{home_dir}/notes.txt", content="Hello from AIO Sandbox!\nLine two.\n", encoding="utf-8", ) print(result.data.bytes_written) # 34 # Write a binary file (base64-encoded PNG) with open("image.png", "rb") as f: b64 = base64.b64encode(f.read()).decode("utf-8") client.file.write_file( file=f"{home_dir}/image.png", content=b64, encoding="base64", ) # Append to an existing file client.file.write_file( file=f"{home_dir}/notes.txt", content="Appended line.\n", append=True, ) ``` --- ## file.read_file / file.readFile Read file contents from the sandbox. Supports line-range selection and optional sudo access. ```python from agent_sandbox import Sandbox client = Sandbox(base_url="http://localhost:8080") # Read entire file result = client.file.read_file(file="/home/gem/notes.txt") print(result.data.content) # Hello from AIO Sandbox!\nLine two.\n # Read a specific line range (0-indexed) result = client.file.read_file( file="/home/gem/notes.txt", start_line=0, end_line=1, ) print(result.data.content) ``` ```typescript import { SandboxClient } from '@agent-infra/sandbox'; const client = new SandboxClient({ environment: 'http://localhost:8080' }); const res = await client.file.readFile({ file: '/tmp/example.txt' }); if (res.ok) console.log(res.body.data.content); ``` --- ## file.list_path / file.listPath List directory contents with metadata: name, size, extension, and whether each entry is a directory. ```python from agent_sandbox import Sandbox client = Sandbox(base_url="http://localhost:8080") result = client.file.list_path(path="/home/gem") print(f"Total items: {result.data.total_count}") for f in result.data.files: kind = "DIR" if f.is_directory else "FILE" print(f" [{kind}] {f.name} ({f.size} bytes)") # Recursive listing result = client.file.list_path(path="/home/gem", recursive=True) ``` --- ## file.search_in_file / file.searchInFile Search for a regex pattern within a single file and return matching lines with line numbers. ```python from agent_sandbox import Sandbox client = Sandbox(base_url="http://localhost:8080") result = client.file.search_in_file( file="/home/gem/app.py", regex=r"def \w+", ) for match in result.data.matches: print(f"Line {match.line_number}: {match.line}") ``` --- ## file.grep_files / file.grepFiles Search across multiple files using a regex pattern, with include/exclude glob filters. ```python from agent_sandbox import Sandbox client = Sandbox(base_url="http://localhost:8080") result = client.file.grep_files( path="/home/gem/project", pattern="TODO", include=["*.py", "*.ts"], exclude=["node_modules", "__pycache__"], case_insensitive=True, recursive=True, max_results=50, ) for match in result.data.matches: print(f"{match.file}:{match.line_number} {match.line.strip()}") ``` --- ## file.find_files / file.findFiles Find files under a directory matching a glob pattern. ```python from agent_sandbox import Sandbox client = Sandbox(base_url="http://localhost:8080") result = client.file.find_files(path="/home/gem", glob="**/*.py") for path in result.data.files: print(path) ``` --- ## file.replace_in_file / file.replaceInFile Replace an exact string occurrence inside a file in place. ```python from agent_sandbox import Sandbox client = Sandbox(base_url="http://localhost:8080") # Replace first occurrence client.file.replace_in_file( file="/home/gem/config.py", old_str='DEBUG = False', new_str='DEBUG = True', ) ``` --- ## file.str_replace_editor Advanced editor tool supporting `view`, `create`, `str_replace`, `insert`, and `undo_edit` operations on files, mirroring a text-editor command set. ```python from agent_sandbox import Sandbox client = Sandbox(base_url="http://localhost:8080") # Create a new file client.file.str_replace_editor( command="create", path="/home/gem/hello.py", file_text="def greet(name):\n return f'Hello, {name}!'\n", ) # View the file result = client.file.str_replace_editor( command="view", path="/home/gem/hello.py", ) print(result.data.content) # Replace a string client.file.str_replace_editor( command="str_replace", path="/home/gem/hello.py", old_str="Hello, {name}!", new_str="Hi, {name}!", ) # Insert a line after line 1 client.file.str_replace_editor( command="insert", path="/home/gem/hello.py", insert_line=1, new_str=' """Greet a user."""\n', ) # Undo the last edit client.file.str_replace_editor( command="undo_edit", path="/home/gem/hello.py", ) ``` --- ## jupyter.create_session and jupyter.execute_code Create a persistent Jupyter kernel session and execute Python code inside it. Variables and imports persist across calls within the same session. ```python from agent_sandbox import Sandbox client = Sandbox(base_url="http://localhost:8080") # Create a kernel session session = client.jupyter.create_session(kernel_name="python3") sid = session.data.session_id # Execute code (state is shared across calls) client.jupyter.execute_code( code="import pandas as pd\ndf = pd.DataFrame({'x': [1,2,3], 'y': [4,5,6]})", session_id=sid, kernel_name="python3", ) result = client.jupyter.execute_code( code="print(df.describe())", session_id=sid, kernel_name="python3", ) print(result.data.outputs[0].text) # Install a package mid-session, then import client.shell.exec_command(command="pip install requests -q") result = client.jupyter.execute_code( code="import requests; print(requests.__version__)", session_id=sid, kernel_name="python3", ) print(result.data.outputs[0].text) # Delete session when done client.jupyter.delete_session(sid) ``` --- ## nodejs.execute_code / nodejs.executeCode Execute Node.js code in the sandbox and capture stdout. ```python from agent_sandbox import Sandbox client = Sandbox(base_url="http://localhost:8080") result = client.nodejs.execute_code(code=""" const os = require('os'); console.log('Platform:', os.platform()); console.log('Node version:', process.version); console.log('Sum 1-100:', Array.from({length:100},(_,i)=>i+1).reduce((a,b)=>a+b,0)); """) print(result.data.outputs[0].text) # Platform: linux # Node version: v20.x.x # Sum 1-100: 5050 ``` ```typescript import { SandboxClient } from '@agent-infra/sandbox'; const client = new SandboxClient({ environment: 'http://localhost:8080' }); const res = await client.nodejs.executeCode({ code: ` const os = require('os'); console.log('Platform:', os.platform()); console.log('Node version:', process.version); `, }); if (res.ok) console.log(res.body.data.outputs); ``` --- ## code.execute_code / code.executeCode Execute Python or JavaScript code through a unified interface without needing an explicit Jupyter session. ```python from agent_sandbox import Sandbox client = Sandbox(base_url="http://localhost:8080") # Python py_result = client.code.execute_code( code="import sys, json\nprint(json.dumps({'version': sys.version.split()[0], 'sum': sum(range(100))}))", language="python", ) print(py_result.data.outputs[0].text) # {"version": "3.12.x", "sum": 4950} ``` ```typescript import { SandboxClient } from '@agent-infra/sandbox'; const client = new SandboxClient({ environment: 'http://localhost:8080' }); // Python const py = await client.code.executeCode({ language: 'python', code: 'print("Hello from Python!")\nprint(2 + 2)', }); if (py.ok) console.log(py.body); // JavaScript const js = await client.code.executeCode({ language: 'javascript', code: 'const arr = [1,2,3,4,5]; console.log(JSON.stringify(arr.map(x => x*x)));', }); if (js.ok) console.log(js.body); ``` --- ## browser.get_info and browser.screenshot Retrieve browser metadata (CDP URL, viewport dimensions) and take a screenshot. ```python from agent_sandbox import Sandbox client = Sandbox(base_url="http://localhost:8080") # Get browser info including CDP WebSocket URL info = client.browser.get_info().data print(info.cdp_url) # ws://localhost:... print(info.viewport) # {"width": 1920, "height": 1080} # Take a screenshot (returns raw bytes) screenshot = client.browser.screenshot() with open("screen.png", "wb") as f: f.write(screenshot.data) # Configure resolution client.browser.set_config(resolution="1920x1080") ``` --- ## browser_page — Navigation and Page Interaction `browser_page` provides fine-grained page control: navigate, click, fill, evaluate JavaScript, scroll, type text, and capture screenshots of individual pages. ```python from agent_sandbox import Sandbox client = Sandbox(base_url="http://localhost:8080") # Navigate to a URL client.browser_page.navigate(url="https://example.com", wait_until="networkidle") # Get page text text = client.browser_page.get_text() print(text.data.text[:200]) # Get page markdown md = client.browser_page.get_markdown() print(md.data.content[:200]) # Evaluate JavaScript on the page result = client.browser_page.evaluate(script="document.title") print(result.data.result) # Example Domain # Click an element by CSS selector client.browser_page.click(selector="#submit-button") # Fill an input client.browser_page.fill(selector='input[name="q"]', value="AIO Sandbox") # Press a key client.browser_page.press_key(key="Enter") # Scroll the page client.browser_page.scroll(amount=1000) # Take a page screenshot screenshot = client.browser_page.screenshot() with open("page.png", "wb") as f: f.write(screenshot.data) ``` --- ## browser_tabs — Tab Management Create, list, switch, and close browser tabs. ```python from agent_sandbox import Sandbox client = Sandbox(base_url="http://localhost:8080") # List all open tabs tabs = client.browser_tabs.list() print(f"Open tabs: {len(tabs.data.tabs)}") # Open a new tab and navigate new_tab = client.browser_tabs.new(url="https://github.com") print(new_tab.data.tab_id) # Switch to a tab by index client.browser_tabs.switch(index=0) # Close a tab by index client.browser_tabs.close(index=1) ``` --- ## browser_cookies — Cookie Operations Read, set, and clear browser cookies for session management. ```python from agent_sandbox import Sandbox client = Sandbox(base_url="http://localhost:8080") # Get all cookies for a URL cookies = client.browser_cookies.get_cookies(url="https://example.com") for cookie in cookies.data.cookies: print(f"{cookie.name}={cookie.value}") # Set a cookie client.browser_cookies.set_cookies(cookies=[{ "name": "session_id", "value": "abc123", "domain": "example.com", "path": "/", }]) # Clear all cookies client.browser_cookies.clear_cookies() ``` --- ## browser_network — Network Monitoring and Routing Inspect HTTP requests, export HAR files, and add/remove request routing rules. ```python from agent_sandbox import Sandbox client = Sandbox(base_url="http://localhost:8080") # Get all captured network requests requests = client.browser_network.get_requests() for req in requests.data.requests: print(f"{req.method} {req.url} → {req.status}") # Export HAR archive har = client.browser_network.export_har() with open("trace.har", "w") as f: f.write(har.data.content) # Add a request route (intercept/mock) client.browser_network.add_route( pattern="**/api/data", response_body='{"mocked": true}', response_status=200, ) # Set custom request headers client.browser_network.set_headers(headers={"X-Custom-Header": "test"}) # Remove a route client.browser_network.remove_route(pattern="**/api/data") ``` --- ## browser_state — Save and Restore Browser State Persist and restore the full browser state (cookies, local storage, session storage) for reproducible sessions. ```python from agent_sandbox import Sandbox client = Sandbox(base_url="http://localhost:8080") # Save current browser state to a file in the sandbox client.browser_state.save(path="/home/gem/browser_state.json") # Later: restore from saved state client.browser_state.load(path="/home/gem/browser_state.json") ``` --- ## Playwright Integration (CDP) Connect Playwright directly to the sandbox browser via CDP for full Playwright API access. ```python import asyncio import os from agent_sandbox import Sandbox from playwright.async_api import async_playwright sandbox = Sandbox(base_url=os.getenv("SANDBOX_BASE_URL", "http://localhost:8080")) cdp_url = sandbox.browser.get_info().data.cdp_url async def main(): async with async_playwright() as pw: # Connect to the sandbox browser over CDP browser = await pw.chromium.connect_over_cdp(cdp_url) context = await browser.new_context() page = await context.new_page() await page.goto("https://duckduckgo.com") search_box = page.locator('input[name="q"]') await search_box.fill("playwright automation") await search_box.press("Enter") await page.wait_for_selector('[data-testid="result"]', timeout=10000) first = page.locator('[data-testid="result"]').first title = await first.locator('h2').inner_text() link = await first.locator('a').get_attribute('href') print(f"Title: {title}\nLink: {link}") await page.screenshot(path="results.png") await context.close() await browser.close() asyncio.run(main()) ``` --- ## MCP Hub — HTTP Endpoint The sandbox exposes a pre-configured MCP Hub at `/mcp` aggregating Browser, File, Terminal, and Markitdown servers. ```python import httpx import asyncio async def call_mcp_tool(tool_name: str, **kwargs): async with httpx.AsyncClient() as client: response = await client.post( "http://localhost:8080/mcp", json={ "method": "tools/call", "params": {"name": tool_name, "arguments": kwargs}, }, timeout=30.0, ) response.raise_for_status() result = response.json() if "error" in result: raise RuntimeError(f"Tool error: {result['error']}") return result async def main(): # Navigate browser and extract content await call_mcp_tool("browser_navigate", url="https://example.com") content = await call_mcp_tool("browser_extract") # Save extracted content to file await call_mcp_tool("file_write", path="/tmp/content.txt", content=content["text"]) # Read it back data = await call_mcp_tool("file_read", path="/tmp/content.txt") print(data) # Execute a shell command result = await call_mcp_tool("terminal_execute", command="uname -a") print(result) asyncio.run(main()) ``` Available MCP tools: `browser_navigate`, `browser_click`, `browser_type`, `browser_screenshot`, `browser_extract`, `file_read`, `file_write`, `file_list`, `file_search`, `file_replace`, `terminal_execute`, `terminal_session`, `terminal_kill`, `markitdown_convert`, `markitdown_extract`. --- ## Authentication — JWT and Short-Lived Tickets Secure the sandbox with RSA key-pair JWT authentication; use tickets for VNC or other header-less connections. ```bash # 1. Generate key pair openssl genrsa -out private_key.pem 2048 openssl rsa -in private_key.pem -pubout -out public_key.pem # 2. Start sandbox with public key export JWT_PUBLIC_KEY=$(cat public_key.pem | base64) docker run --security-opt seccomp=unconfined -p 8080:8080 \ -e JWT_PUBLIC_KEY="${JWT_PUBLIC_KEY}" \ ghcr.io/agent-infra/sandbox:latest # 3. Generate a JWT with the private key (simplified — use a JWT library in production) base64url_encode() { openssl base64 -e -A | tr '+/' '-_' | tr -d '='; } header='{"alg":"RS256","typ":"JWT"}' exp_time=$(($(date +%s) + 3600)) payload="{\"exp\":${exp_time}}" to_be_signed="$(echo -n "$header" | base64url_encode).$(echo -n "$payload" | base64url_encode)" signature=$(echo -n "$to_be_signed" | openssl dgst -sha256 -sign private_key.pem | base64url_encode) jwt="${to_be_signed}.${signature}" # 4. Call the API with Bearer token curl -X GET "http://localhost:8080/v1/sandbox" -H "Authorization: Bearer ${jwt}" # 5. Exchange JWT for a short-lived ticket (for VNC / WebSocket URLs) ticket_response=$(curl -X POST "http://localhost:8080/tickets" \ -H "Authorization: Bearer ${jwt}") ticket=$(echo "$ticket_response" | jq -r .ticket) # 6. Use ticket in browser URL (no Authorization header needed) echo "http://localhost:8080/vnc/index.html?ticket=${ticket}&path=websockify%3Fticket%3D${ticket}" ``` --- ## Docker Compose Deployment Full production deployment configuration with environment-variable driven options. ```yaml # docker-compose.yaml version: '3.8' services: sandbox: container_name: aio-sandbox image: ghcr.io/agent-infra/sandbox:latest volumes: - /tmp/gem/project:/home/gem/project security_opt: - seccomp:unconfined extra_hosts: - "host.docker.internal:host-gateway" restart: "unless-stopped" shm_size: "2gb" ports: - "${HOST_PORT:-8080}:8080" environment: PROXY_SERVER: ${PROXY_SERVER:-host.docker.internal:7890} JWT_PUBLIC_KEY: ${JWT_PUBLIC_KEY:-} DNS_OVER_HTTPS_TEMPLATES: ${DNS_OVER_HTTPS_TEMPLATES:-} WORKSPACE: ${WORKSPACE:-"/home/gem"} HOMEPAGE: ${HOMEPAGE:-} BROWSER_EXTRA_ARGS: ${BROWSER_EXTRA_ARGS:-} TZ: ${TZ:-Asia/Singapore} WAIT_PORTS: ${WAIT_PORTS:-} ``` --- ## Volcengine Cloud Provider — Sandbox Lifecycle Management The `VolcengineProvider` manages sandbox instances on Volcengine VEFAAS (create, list, get, delete). ```python import time, os from agent_sandbox import Sandbox from agent_sandbox.providers import VolcengineProvider provider = VolcengineProvider( access_key=os.environ["VOLCENGINE_ACCESS_KEY"], secret_key=os.environ["VOLCENGINE_SECRET_KEY"], region=os.getenv("VOLCENGINE_REGION", "cn-beijing"), ) # Create an application (one-time setup) app_id = provider.create_application(name="my-sandbox-app", gateway_name="my-gateway") # Poll until the application is ready ready, function_id = False, None while not ready: ready, function_id = provider.get_application_readiness(id=app_id) if not ready: time.sleep(1) # Create a sandbox instance sandbox_id = provider.create_sandbox(function_id=function_id) print(f"Sandbox created: {sandbox_id}") # List all sandboxes sandboxes = provider.list_sandboxes(function_id=function_id) print(f"Total sandboxes: {len(sandboxes)}") # Get sandbox details and extract public domain info = provider.get_sandbox(function_id=function_id, sandbox_id=sandbox_id) public_domain = next(d["domain"] for d in info["domains"] if d["type"] == "public") # Use the sandbox via its public URL client = Sandbox(base_url=public_domain) result = client.file.list_path(path=".") print(result) # Delete the sandbox when done provider.delete_sandbox(function_id=function_id, sandbox_id=sandbox_id) ``` TypeScript equivalent using `providers.VolcengineProvider`: ```typescript import { SandboxClient, providers } from '@agent-infra/sandbox'; const volcengine = new providers.VolcengineProvider({ accessKey: process.env.VOLCENGINE_ACCESS_KEY!, secretKey: process.env.VOLCENGINE_SECRET_KEY!, region: 'cn-beijing', }); const sandboxId = await volcengine.createSandbox('my-function-id', 60); const info = await volcengine.getSandbox('my-function-id', sandboxId); const sandboxes = await volcengine.listSandboxes('my-function-id'); await volcengine.deleteSandbox('my-function-id', sandboxId); ``` --- ## OpenAI Tool-Calling Integration Expose sandbox execution as an OpenAI function tool, letting an LLM run code and read results in a conversation loop. ```python import json from openai import OpenAI from agent_sandbox import Sandbox client = OpenAI() sandbox = Sandbox(base_url="http://localhost:8080") def run_code(code: str, lang: str = "python") -> str: if lang == "python": result = sandbox.jupyter.execute_code(code=code).data else: result = sandbox.nodejs.execute_code(code=code).data outputs = getattr(result, "outputs", []) return outputs[0].text if outputs else "(no output)" tools = [{ "type": "function", "function": { "name": "run_code", "description": "Execute Python or JavaScript code in a sandbox", "parameters": { "type": "object", "properties": { "code": {"type": "string", "description": "Code to execute"}, "lang": {"type": "string", "enum": ["python", "javascript"], "default": "python"}, }, "required": ["code"], }, }, }] response = client.chat.completions.create( model="gpt-4o", messages=[{"role": "user", "content": "Compute the first 10 Fibonacci numbers and print them."}], tools=tools, ) msg = response.choices[0].message if msg.tool_calls: for call in msg.tool_calls: args = json.loads(call.function.arguments) output = run_code(**args) print(f"Code output:\n{output}") # Expected: 0 1 1 2 3 5 8 13 21 34 ``` --- ## AG2 Multi-Agent Integration Wire AIO Sandbox tools into an AG2 (AutoGen) multi-agent GroupChat where one agent writes code and another reviews results. ```python import uuid from agent_sandbox import Sandbox from autogen import AssistantAgent, GroupChat, GroupChatManager, LLMConfig, UserProxyAgent sandbox = Sandbox(base_url="http://localhost:8080") jupyter_session = sandbox.jupyter.create_session(kernel_name="python3") shell_session = sandbox.shell.create_session(id=str(uuid.uuid4())) j_sid = jupyter_session.data.session_id s_sid = shell_session.data.session_id llm_config = LLMConfig({"model": "gpt-4o-mini", "api_key": "sk-..."}) proxy = UserProxyAgent(name="user_proxy", human_input_mode="NEVER", code_execution_config=False, is_termination_msg=lambda m: "TERMINATE" in (m.get("content") or "")) coder = AssistantAgent(name="coder", system_message="Write and run Python code using run_python.", llm_config=llm_config) reviewer = AssistantAgent(name="reviewer", system_message="Verify results. Say TERMINATE when done.", llm_config=llm_config) @proxy.register_for_execution() @coder.register_for_llm(description="Run Python in a persistent Jupyter kernel") def run_python(code: str) -> str: result = sandbox.jupyter.execute_code(code=code, session_id=j_sid) outputs = getattr(result.data, "outputs", []) return "".join(o.text for o in outputs if hasattr(o, "text")) or "OK" @proxy.register_for_execution() @coder.register_for_llm(description="Run a shell command in the sandbox") def run_shell(command: str) -> str: result = sandbox.shell.exec_command(command=command, id=s_sid) return getattr(result.data, "output", "") or "OK" group_chat = GroupChat(agents=[proxy, coder, reviewer], messages=[], max_round=15) manager = GroupChatManager(groupchat=group_chat, llm_config=llm_config, is_termination_msg=lambda m: "TERMINATE" in (m.get("content") or "")) proxy.run(manager, message="Load the Iris dataset and print summary statistics.").process() # Cleanup sandbox.jupyter.delete_session(j_sid) sandbox.shell.cleanup_session(s_sid) ``` --- ## Site-to-Markdown Unified Workflow Demonstrate the unified filesystem: browser captures HTML, Jupyter converts it, shell lists files, and the file API retrieves the result — all in one sandbox. ```python import asyncio, base64 from playwright.async_api import async_playwright from agent_sandbox import Sandbox async def site_to_markdown(url: str) -> str: c = Sandbox(base_url="http://localhost:8080") home_dir = c.sandbox.get_context().home_dir # 1. Browser: capture HTML and screenshot via CDP async with async_playwright() as p: cdp_url = c.browser.get_info().data.cdp_url browser = await p.chromium.connect_over_cdp(cdp_url) page = await (await browser.new_context()).new_page() await page.goto(url, wait_until="networkidle") html = await page.content() screenshot_b64 = base64.b64encode(await page.screenshot()).decode() await browser.close() # 2. Jupyter: convert HTML → markdown inside the sandbox c.jupyter.execute_code(code=f""" from markdownify import markdownify html = '''{html}''' screenshot_b64 = "{screenshot_b64}" md = f"{{markdownify(html)}}\\n\\n" with open('{home_dir}/site.md', 'w') as f: f.write(md) print("Conversion complete") """) # 3. Shell: verify the file exists listing = c.shell.exec_command(command=f"ls -lh {home_dir}/*.md") print(listing.data.output) # 4. File: read and return the markdown return c.file.read_file(file=f"{home_dir}/site.md").data.content result = asyncio.run(site_to_markdown("https://example.com")) print(result[:500]) ``` --- ## Error Handling and Retry Patterns (TypeScript) Use exponential backoff retries and a circuit-breaker pattern when calling sandbox APIs in production. ```typescript import { SandboxClient } from '@agent-infra/sandbox'; const client = new SandboxClient({ environment: 'http://localhost:8080', timeoutInSeconds: 5, }); // Exponential backoff retry async function retryWithBackoff<T>( fn: () => Promise<T>, maxRetries = 3, initialDelay = 1000, backoffMultiplier = 2, ): Promise<T> { let lastError: Error; for (let attempt = 0; attempt < maxRetries; attempt++) { try { return await fn(); } catch (err) { lastError = err instanceof Error ? err : new Error(String(err)); if (lastError.message.includes('401')) throw new Error('Auth failed — not retrying'); if (attempt < maxRetries - 1) { const delay = Math.min(initialDelay * Math.pow(backoffMultiplier, attempt), 10000); await new Promise(r => setTimeout(r, delay)); } } } throw lastError!; } // Circuit breaker class CircuitBreaker { private failures = 0; private lastFailure = 0; private state: 'CLOSED' | 'OPEN' | 'HALF_OPEN' = 'CLOSED'; constructor(private threshold = 5, private timeout = 60_000) {} async execute<T>(fn: () => Promise<T>): Promise<T> { if (this.state === 'OPEN') { if (Date.now() - this.lastFailure > this.timeout) this.state = 'HALF_OPEN'; else throw new Error('Circuit breaker OPEN'); } try { const result = await fn(); this.state = 'CLOSED'; this.failures = 0; return result; } catch (err) { this.failures++; this.lastFailure = Date.now(); if (this.failures >= this.threshold) this.state = 'OPEN'; throw err; } } } // Usage const breaker = new CircuitBreaker(3, 5000); const result = await retryWithBackoff(() => breaker.execute(() => client.shell.execCommand({ command: 'echo "ok"' })) ); console.log(result.ok ? result.body.data.output : 'failed'); ``` --- AIO Sandbox's primary use cases are AI agent development and automated workflows: an LLM agent can write code, execute it securely in the Jupyter or shell environment, interact with external websites through the sandboxed browser, manipulate files, and receive structured results — all without any local execution risk. The unified filesystem eliminates the file-sharing friction of single-purpose sandboxes, enabling seamless multi-step pipelines such as downloading a file via browser, processing it with Python, and exposing the result via file operations in a single coherent session. For integration, the recommended pattern is to instantiate a `Sandbox` (Python) or `SandboxClient` (TypeScript) client once per agent run, reuse persistent Jupyter and shell sessions for stateful multi-step code execution, and pass the CDP URL to Playwright or Browser Use for rich browser automation. For cloud deployments at scale, `VolcengineProvider` handles the full sandbox instance lifecycle, while JWT authentication and short-lived ticket exchange secure both REST and WebSocket endpoints. The MCP Hub at `/mcp` provides a protocol-standard alternative for AI frameworks that speak MCP natively.