baponi 0.2.1

Sandboxed code execution for AI agents

Baponi Python SDK

Sandboxed code execution for AI agents. Run Bash, Python, and Node.js in secure, isolated containers with sub-20ms overhead.

Installation

pip install baponi

With framework integrations:

pip install baponi[langchain]     # LangChain
pip install baponi[openai]        # OpenAI Agents SDK
pip install baponi[anthropic]     # Anthropic
pip install baponi[google]        # Google Gemini
pip install baponi[crewai]        # CrewAI
pip install baponi[all]           # All frameworks

Quick Start

from baponi import Baponi

client = Baponi()  # reads BAPONI_API_KEY from env
result = client.execute("print('Hello!')")
print(result.stdout)  # Hello!

Async

from baponi import AsyncBaponi

async with AsyncBaponi() as client:
    result = await client.execute("print('Hello!')")
    print(result.stdout)

Supported Languages

client.execute("echo 'Bash'", language="bash")
client.execute("print('Python')")
client.execute("console.log('Node')", language="node")

Persistent State

Pass a thread_id to persist files and installed packages across calls:

client.execute("pip install pandas", language="bash", thread_id="analysis-session")
client.execute("""
import pandas as pd
df = pd.DataFrame({'x': [1, 2, 3]})
df.to_csv('/home/baponi/data.csv', index=False)
print(df.describe())
""", thread_id="analysis-session")

Framework Integrations

LangChain

from baponi.langchain import code_sandbox
from langchain.agents import create_react_agent

agent = create_react_agent(llm, tools=[code_sandbox])

OpenAI Agents SDK

from baponi.openai import code_sandbox
from agents import Agent

agent = Agent(name="coder", tools=[code_sandbox])

Anthropic

from baponi.anthropic import code_sandbox_tool, handle_tool_call
import anthropic

client = anthropic.Anthropic()
response = client.messages.create(
    model="claude-sonnet-4-20250514",
    tools=[code_sandbox_tool],
    messages=[{"role": "user", "content": "Calculate fibonacci(10) in Python"}],
)

for block in response.content:
    if block.type == "tool_use":
        result = handle_tool_call(block.name, block.input)
        print(result)

Google Gemini

from baponi.google import code_sandbox
from google import genai

client = genai.Client()
chat = client.chats.create(
    model="gemini-2.5-flash",
    config={"tools": [code_sandbox]},
)
response = chat.send_message("Calculate pi to 100 digits")

CrewAI

from baponi.crewai import code_sandbox
from crewai import Agent

agent = Agent(role="Data Analyst", tools=[code_sandbox])

Custom Configuration

All integrations support create_code_sandbox() for power users:

from baponi.langchain import create_code_sandbox

sandbox = create_code_sandbox(
    api_key="sk-...",
    base_url="https://your-baponi-instance.com",
    thread_id="shared-session",        # Default thread for all calls
    timeout=120,                        # Default timeout
    metadata={"user_id": "usr_123"},   # Metadata on every call
)

Error Handling

API errors and execution errors are separate concepts:

from baponi import Baponi, AuthenticationError, RateLimitError

client = Baponi()

# API errors raise exceptions
try:
    result = client.execute("print(1)")
except AuthenticationError:
    print("Invalid API key")
except RateLimitError as e:
    print(f"Rate limited, retry after {e.retry_after}s")

# Execution errors return SandboxResult with success=False
result = client.execute("raise ValueError('oops')")
if not result.success:
    print(f"Code failed with exit code {result.exit_code}")
    print(f"stderr: {result.stderr}")

Exception Hierarchy

Exception	HTTP Status	Description
BaponiError	—	Base exception for all API errors
AuthenticationError	401	Invalid or missing API key
ForbiddenError	403	Insufficient permissions
RateLimitError	429	Rate limit exceeded
ThreadBusyError	409	Thread already executing
APITimeoutError	504	Server-side timeout
ServerError	500/503	Server error
APIValidationError	400	Invalid request

Configuration

from baponi import Baponi

# Self-hosted deployment
client = Baponi(
    api_key="sk-...",
    base_url="https://baponi.internal.company.com",
)

# Custom HTTP client (proxies, observability, custom TLS)
import httpx

http_client = httpx.Client(
    proxies="http://proxy.internal:8080",
    verify="/path/to/ca-bundle.crt",
)
client = Baponi(api_key="sk-...", http_client=http_client)

# Retry configuration
client = Baponi(
    api_key="sk-...",
    max_retries=0,    # Disable retries
    timeout=120.0,    # Connection timeout (not execution timeout)
)

SandboxResult

result = client.execute("print('hi')")

result.success              # bool — True if exit_code == 0
result.stdout               # str — standard output
result.stderr               # str — standard error
result.exit_code            # int — process exit code
result.network_egress_bytes # int — bytes sent to network
result.storage_egress_bytes # int — bytes written to storage
result.error                # str | None — error message if failed
result.model_dump()         # dict — Pydantic serialization

Environment Variables

Set environment variables in the sandbox:

result = client.execute(
    "import os; print(os.environ['DATABASE_URL'])",
    env_vars={"DATABASE_URL": "postgres://localhost/mydb", "DEBUG": "true"},
)

Keys must be uppercase (MY_VAR), max 50 variables. System-reserved names (PATH, HOME, etc.) are blocked.

Streaming Execution

Get real-time stdout/stderr as the code runs:

with client.execute_stream("for i in range(5): print(i)") as stream:
    for event in stream:
        if isinstance(event, baponi.OutputEvent):
            print(event.data, end="")
    result = stream.get_final_result()

Or just consume silently:

with client.execute_stream("print('hello')") as stream:
    result = stream.until_done()

Async:

async with await client.execute_stream("print('hi')") as stream:
    async for event in stream:
        print(event)

Event types: StatusEvent, OutputEvent, KeepaliveEvent, ResultEvent.

Webhook (Async) Execution

Fire-and-forget execution. The server accepts immediately and optionally POSTs the result to a webhook URL:

handle = client.execute_webhook(
    "import time; time.sleep(10); print('done')",
    webhook_url="https://example.com/webhook",  # optional
)
print(handle.trace_id)  # immediately available

# Poll for status
status = handle.poll()

# Or wait until completion
status = handle.wait(poll_interval=2.0, timeout=60.0)
print(status.status)  # "success"

Status Checking & Cancellation

Check execution status or cancel a running execution by trace_id:

status = client.get_execution("trc_abc12345")
print(status.status)

cancel = client.cancel_execution("trc_abc12345")
print(cancel.status)  # "cancelling" — poll to confirm "cancelled"

Coming Soon

• Web Tools API — web search and fetch from within sandboxes (/v1/web/search, /v1/web/fetch)