API reference¶
Generated from the apogee-ai-sandbox source with mkdocstrings. Every symbol below is exported from apogee_ai_sandbox, so it is part of the supported public surface.
Application¶
Python
build_request(dto: RunCodeDTO) -> ExecutionRequest
Source code in apogee_ai_sandbox/application/services/request_builder.py
Python
def build_request(dto: RunCodeDTO) -> ExecutionRequest:
files = tuple(SandboxFile(path=path, content=content) for path, content in dto.files.items())
quotas = Quotas(
cpu_seconds=dto.cpu_seconds,
memory_mb=dto.memory_mb,
wall_seconds=dto.wall_seconds,
pids=dto.pids,
)
egress = EgressAllowlist(policy=dto.network, hosts=tuple(dto.egress_hosts))
return ExecutionRequest(
code=dto.code,
language=dto.language,
files=files,
inputs=dict(dto.inputs),
quotas=quotas,
egress=egress,
env=dict(dto.env),
tenant_id=dto.tenant_id,
user_id=dto.user_id,
metadata=dict(dto.metadata),
cache=dto.cache,
)
Application · DTOs¶
Bases: BaseModel
Application · Use cases¶
Python
KillJobUseCase(sandbox: ISandbox)
Source code in apogee_ai_sandbox/application/use_cases/kill_job_use_case.py
async
¶
Python
ListBackendsUseCase(registry: SandboxRegistry)
Source code in apogee_ai_sandbox/application/use_cases/list_backends_use_case.py
Source code in apogee_ai_sandbox/application/use_cases/list_backends_use_case.py
Python
RunCodeStreamUseCase(sandbox: ISandbox)
Source code in apogee_ai_sandbox/application/use_cases/run_code_stream_use_case.py
async
¶
Python
execute(request: ExecutionRequest) -> AsyncIterator[StreamChunk]
Python
RunCodeUseCase(sandbox: ISandbox, *, cache: IFingerprintCache | None = None, quotas: IQuotaEnforcer | None = None, egress_gate: EgressAllowlistGate | None = None)
Orchestrates: egress gate → quota → cache → execute → release.
Source code in apogee_ai_sandbox/application/use_cases/run_code_use_case.py
Python
def __init__(
self,
sandbox: ISandbox,
*,
cache: IFingerprintCache | None = None,
quotas: IQuotaEnforcer | None = None,
egress_gate: EgressAllowlistGate | None = None,
) -> None:
self._sandbox = sandbox
self._cache = cache
self._quotas = quotas
self._egress_gate = egress_gate or EgressAllowlistGate()
async
¶
Python
execute(request: ExecutionRequest) -> ExecutionResult
Source code in apogee_ai_sandbox/application/use_cases/run_code_use_case.py
Python
async def execute(self, request: ExecutionRequest) -> ExecutionResult:
try:
self._egress_gate.check(code=request.code, allowlist=request.egress)
except EgressBlockedException as exc:
return ExecutionResult(
job_id="",
backend=self._sandbox.name,
status=JobStatus.EGRESS_BLOCKED,
error=str(exc),
)
cache_key: OutputCacheKey | None = None
if request.cache and self._cache is not None:
fingerprint = Fingerprint.of(
code=request.code,
language=request.language.value,
inputs=request.inputs,
files={f.path: f.content for f in request.files},
)
cache_key = OutputCacheKey(fingerprint=fingerprint, backend=self._sandbox.name)
cached = await self._cache.get(cache_key)
if cached is not None:
return cached
if self._quotas is not None:
try:
granted = await self._quotas.acquire(
tenant_id=request.tenant_id, quotas=request.quotas
)
except QuotaExceededException as exc:
return ExecutionResult(
job_id="",
backend=self._sandbox.name,
status=JobStatus.QUOTA_EXCEEDED,
error=str(exc),
)
if not granted:
return ExecutionResult(
job_id="",
backend=self._sandbox.name,
status=JobStatus.QUOTA_EXCEEDED,
error="quota acquire returned False",
)
observed_cpu = 0.0
try:
start = time.perf_counter()
result = await self._sandbox.run(request)
observed_cpu = (time.perf_counter() - start)
if cache_key is not None and self._cache is not None and result.succeeded:
await self._cache.put(cache_key, result)
return result
except SandboxError as exc:
return ExecutionResult(
job_id="",
backend=self._sandbox.name,
status=JobStatus.FAILED,
error=str(exc),
)
finally:
if self._quotas is not None:
await self._quotas.release(
tenant_id=request.tenant_id, observed_cpu_seconds=observed_cpu
)
Domain¶
dataclass
¶
Python
EgressAllowlist(policy: NetworkPolicy = NONE, hosts: tuple[str, ...] = tuple())
Network egress policy.
hosts accepts exact hosts, IPv4 CIDRs and fnmatch patterns
(*.example.com). policy decides the default behaviour when the
allowlist is empty.
class-attribute
instance-attribute
¶
Source code in apogee_ai_sandbox/domain/value_objects/egress_allowlist.py
Python
def is_allowed(self, host: str) -> bool:
if self.policy == NetworkPolicy.UNRESTRICTED:
return True
if self.policy == NetworkPolicy.NONE:
return False
# ALLOWLIST mode
host = host.strip().lower()
for pattern in self.hosts:
if "*" in pattern or "?" in pattern:
if fnmatch.fnmatch(host, pattern.lower()):
return True
elif host == pattern.lower() or host.endswith("." + pattern.lower()):
return True
return False
dataclass
¶
Python
ExecutionRequest(code: str, language: Language = PYTHON, files: tuple[SandboxFile, ...] = tuple(), inputs: dict[str, Any] = dict(), quotas: Quotas = Quotas(), egress: EgressAllowlist = EgressAllowlist(), env: dict[str, str] = dict(), workdir: str = '/sandbox', tenant_id: str | None = None, user_id: str | None = None, metadata: dict[str, str] = dict(), cache: bool = True)
All inputs required to schedule a sandboxed run.
class-attribute
instance-attribute
¶
Python
files: tuple[SandboxFile, ...] = field(default_factory=tuple)
class-attribute
instance-attribute
¶
class-attribute
instance-attribute
¶
Python
egress: EgressAllowlist = field(default_factory=EgressAllowlist)
class-attribute
instance-attribute
¶
dataclass
¶
Python
ExecutionResult(job_id: str, backend: str, status: JobStatus, stdout: str = '', stderr: str = '', exit_code: int | None = None, duration_ms: float = 0.0, cpu_seconds: float = 0.0, memory_peak_mb: float = 0.0, artifacts: dict[str, str] = dict(), error: str | None = None, cached: bool = False)
Final outcome of an execution job.
class-attribute
instance-attribute
¶
Optional named outputs (file path → contents/base64).
dataclass
¶
Deterministic content-addressable id for code+inputs.
classmethod
¶
Python
of(*, code: str, language: str, inputs: dict[str, Any] | None = None, files: dict[str, str] | None = None) -> Fingerprint
Source code in apogee_ai_sandbox/domain/value_objects/fingerprint.py
Python
@classmethod
def of(
cls,
*,
code: str,
language: str,
inputs: dict[str, Any] | None = None,
files: dict[str, str] | None = None,
) -> Fingerprint:
h = hashlib.sha256()
h.update(language.encode())
h.update(b"\x00")
h.update(code.encode())
h.update(b"\x00")
if inputs:
h.update(
json.dumps(inputs, sort_keys=True, separators=(",", ":"), default=str).encode()
)
h.update(b"\x00")
if files:
for name in sorted(files):
h.update(name.encode())
h.update(b"\x00")
h.update(files[name].encode())
h.update(b"\x00")
return cls(digest=h.hexdigest())
Bases: str, Enum
Bases: str, Enum
dataclass
¶
Python
OutputCacheKey(fingerprint: Fingerprint, backend: str)
Composite key used by the IFingerprintCache.
instance-attribute
¶
Backend name — different sandboxes may yield different outputs.
dataclass
¶
Python
Quotas(cpu_seconds: float = 5.0, memory_mb: int = 256, wall_seconds: float = 10.0, pids: int = 64, file_size_mb: int = 32, open_files: int = 64)
Resource limits enforced by the sandbox runtime.
All fields are upper bounds. cpu_seconds is a soft CPU-time cap;
wall_seconds is the hard wall-clock timeout. Backends translate to
cgroups, prlimit, ulimit, container resource flags or cloud quotas.
dataclass
¶
dataclass
¶
Python
SandboxJob(id: str = (lambda: token_hex(8))(), request: ExecutionRequest | None = None, backend: str = '', status: JobStatus = PENDING, started_at: datetime = (lambda: now(utc))(), finished_at: datetime | None = None, result: ExecutionResult | None = None)
Persistent record of a scheduled sandbox execution.
class-attribute
instance-attribute
¶
dataclass
¶
Python
StreamChunk(job_id: str, channel: str, payload: str = '', timestamp: datetime = (lambda: now(utc))())
Single delta in a streaming execution: stdout, stderr or status.
class-attribute
instance-attribute
¶
Domain · Enums¶
Bases: str, Enum
Domain · Exceptions¶
Bases: SandboxError
Bases: SandboxError
Source code in apogee_ai_sandbox/domain/exceptions/sandbox_exceptions.py
Bases: SandboxError
Source code in apogee_ai_sandbox/domain/exceptions/sandbox_exceptions.py
Bases: Exception
Base exception for apogee-ai-sandbox.
Domain · Protocols (ports)¶
Bases: Protocol
async
¶
Python
get(key: OutputCacheKey) -> ExecutionResult | None
async
¶
Python
put(key: OutputCacheKey, result: ExecutionResult) -> None
async
¶
Bases: Protocol
Token-bucket-style quota gate per tenant.
The enforcer checks whether the requested run fits the tenant's window.
Implementations either raise QuotaExceededException or return
False based on the constructor flag.
Bases: Protocol
Backend that runs a single job inside an isolation boundary.
async
¶
Python
run(request: ExecutionRequest) -> ExecutionResult
async
¶
Python
stream(request: ExecutionRequest) -> AsyncIterator[StreamChunk]
async
¶
async
¶
Infrastructure¶
Bases: _JailSandbox
Source code in apogee_ai_sandbox/infrastructure/adapters/jail_sandbox.py
Python
CodeInterpreterTool(sandbox: ISandbox, *, default_quotas: Quotas | None = None, default_egress: EgressAllowlist | None = None)
Bases: SandboxTool
Source code in apogee_ai_sandbox/infrastructure/tools/sandbox_tools.py
Adapter for Daytona dev-environments / sandboxes.
Lazy import: install via pip install 'apogee-ai-sandbox[daytona]'.
Source code in apogee_ai_sandbox/infrastructure/adapters/daytona_sandbox.py
Python
def __init__(self, *, api_key: str | None = None, server_url: str | None = None) -> None:
try:
import daytona_sdk # type: ignore # noqa: F401
except ImportError as exc:
raise ImportError(
"DaytonaSandboxAdapter requires `daytona-sdk`. "
"Install with: pip install 'apogee-ai-sandbox[daytona]'"
) from exc
self._api_key = api_key
self._server_url = server_url
class-attribute
instance-attribute
¶
Python
languages = (PYTHON, JAVASCRIPT, TYPESCRIPT, BASH)
async
¶
Python
run(request: ExecutionRequest) -> ExecutionResult
Source code in apogee_ai_sandbox/infrastructure/adapters/daytona_sandbox.py
Python
async def run(self, request: ExecutionRequest) -> ExecutionResult:
if not self.supports(request.language):
raise LanguageNotSupportedException(request.language.value, self.name)
try:
from daytona_sdk import Daytona # type: ignore
except ImportError as exc: # pragma: no cover
raise ImportError(str(exc)) from exc
job_id = secrets.token_hex(8)
start = time.perf_counter()
def _run() -> ExecutionResult:
client = Daytona(api_key=self._api_key, server_url=self._server_url)
workspace = client.create()
try:
if request.language == Language.PYTHON:
response = workspace.process.code_run(request.code)
else:
response = workspace.process.exec(request.code)
stdout = getattr(response, "result", "") or getattr(response, "stdout", "")
exit_code = int(getattr(response, "exit_code", 0) or 0)
status = JobStatus.SUCCEEDED if exit_code == 0 else JobStatus.FAILED
return ExecutionResult(
job_id=job_id,
backend=self.name,
status=status,
stdout=str(stdout),
exit_code=exit_code,
duration_ms=(time.perf_counter() - start) * 1000.0,
)
finally:
try:
client.remove(workspace)
except Exception: # noqa: BLE001
pass
try:
return await asyncio.wait_for(
asyncio.to_thread(_run), timeout=request.quotas.wall_seconds + 30
)
except TimeoutError:
return ExecutionResult(
job_id=job_id,
backend=self.name,
status=JobStatus.TIMEOUT,
error=f"wall_seconds={request.quotas.wall_seconds} exceeded",
duration_ms=(time.perf_counter() - start) * 1000.0,
)
async
¶
Python
stream(request: ExecutionRequest) -> AsyncIterator[StreamChunk]
Source code in apogee_ai_sandbox/infrastructure/adapters/daytona_sandbox.py
Python
async def stream(self, request: ExecutionRequest) -> AsyncIterator[StreamChunk]:
result = await self.run(request)
async def gen() -> AsyncIterator[StreamChunk]:
if result.stdout:
yield StreamChunk(job_id=result.job_id, channel="stdout", payload=result.stdout)
yield StreamChunk(
job_id=result.job_id, channel="status", payload=result.status.value
)
return gen()
async
¶
async
¶
Python
DockerSandboxAdapter(*, image_overrides: dict[Language, str] | None = None, run_as_user: str = '65534:65534')
Run code inside a single-shot Docker container.
Lazy import: install via pip install 'apogee-ai-sandbox[docker]'.
Honors quotas through --memory, --cpus, --pids-limit and
drops capabilities by default. Network mode follows the egress policy.
Source code in apogee_ai_sandbox/infrastructure/adapters/docker_sandbox.py
Python
def __init__(
self,
*,
image_overrides: dict[Language, str] | None = None,
run_as_user: str = "65534:65534",
) -> None:
try:
import docker # type: ignore # noqa: F401
except ImportError as exc:
raise ImportError(
"DockerSandboxAdapter requires `docker`. "
"Install with: pip install 'apogee-ai-sandbox[docker]'"
) from exc
from docker import from_env # type: ignore
self._client = from_env()
self._image_overrides = dict(image_overrides or {})
self._run_as_user = run_as_user
self._jobs: dict[str, str] = {}
async
¶
Python
run(request: ExecutionRequest) -> ExecutionResult
Source code in apogee_ai_sandbox/infrastructure/adapters/docker_sandbox.py
Python
async def run(self, request: ExecutionRequest) -> ExecutionResult:
if not self.supports(request.language):
raise LanguageNotSupportedException(request.language.value, self.name)
image = self._image_overrides.get(
request.language, _LANGUAGE_DEFAULTS[request.language][0]
)
cmd = list(_LANGUAGE_DEFAULTS[request.language][1]) + [request.code]
network_mode = self._network_mode(request)
cpu_quota = max(10000, int(request.quotas.cpu_seconds * 100000))
job_id = secrets.token_hex(8)
start = time.perf_counter()
def _run() -> tuple[int, bytes, str]:
container = self._client.containers.run(
image=image,
command=cmd,
detach=True,
user=self._run_as_user,
read_only=True,
tmpfs={"/tmp": "rw,size=64m,nosuid,nodev,noexec"},
network_mode=network_mode,
mem_limit=f"{request.quotas.memory_mb}m",
memswap_limit=f"{request.quotas.memory_mb}m",
pids_limit=request.quotas.pids,
cap_drop=["ALL"],
security_opt=["no-new-privileges"],
ulimits=[
{
"Name": "cpu",
"Soft": int(request.quotas.cpu_seconds),
"Hard": int(request.quotas.cpu_seconds),
}
],
cpu_quota=cpu_quota,
environment={k: v for k, v in request.env.items()},
)
self._jobs[job_id] = container.id
try:
exit_status = container.wait(timeout=int(request.quotas.wall_seconds))
exit_code = int(exit_status.get("StatusCode", -1))
logs = container.logs(stdout=True, stderr=True)
finally:
container.remove(force=True)
self._jobs.pop(job_id, None)
return exit_code, logs, container.id
try:
exit_code, logs, _ = await asyncio.to_thread(_run)
except Exception as exc: # noqa: BLE001 - any docker-py error
return ExecutionResult(
job_id=job_id,
backend=self.name,
status=JobStatus.FAILED,
error=str(exc),
duration_ms=(time.perf_counter() - start) * 1000.0,
)
duration_ms = (time.perf_counter() - start) * 1000.0
status = JobStatus.SUCCEEDED if exit_code == 0 else JobStatus.FAILED
return ExecutionResult(
job_id=job_id,
backend=self.name,
status=status,
stdout=logs.decode(errors="replace") if isinstance(logs, bytes) else str(logs),
exit_code=exit_code,
duration_ms=duration_ms,
)
async
¶
Python
stream(request: ExecutionRequest) -> AsyncIterator[StreamChunk]
Source code in apogee_ai_sandbox/infrastructure/adapters/docker_sandbox.py
Python
async def stream(self, request: ExecutionRequest) -> AsyncIterator[StreamChunk]:
result = await self.run(request)
async def gen() -> AsyncIterator[StreamChunk]:
if result.stdout:
yield StreamChunk(job_id=result.job_id, channel="stdout", payload=result.stdout)
yield StreamChunk(
job_id=result.job_id, channel="status", payload=result.status.value
)
return gen()
async
¶
Source code in apogee_ai_sandbox/infrastructure/adapters/docker_sandbox.py
Python
async def kill(self, job_id: str) -> bool:
container_id = self._jobs.get(job_id)
if container_id is None:
return False
def _kill() -> None:
try:
container = self._client.containers.get(container_id)
container.kill()
except Exception: # noqa: BLE001
return
await asyncio.to_thread(_kill)
return True
async
¶
Adapter for E2B Code Interpreter cloud sandboxes.
Lazy import: install via pip install 'apogee-ai-sandbox[e2b]'.
Source code in apogee_ai_sandbox/infrastructure/adapters/e2b_sandbox.py
Python
def __init__(self, *, api_key: str | None = None, template: str = "code-interpreter-v1") -> None:
try:
import e2b_code_interpreter # type: ignore # noqa: F401
except ImportError as exc:
raise ImportError(
"E2BSandboxAdapter requires `e2b-code-interpreter`. "
"Install with: pip install 'apogee-ai-sandbox[e2b]'"
) from exc
self._api_key = api_key
self._template = template
class-attribute
instance-attribute
¶
Python
languages = (PYTHON, JAVASCRIPT, TYPESCRIPT, BASH)
async
¶
Python
run(request: ExecutionRequest) -> ExecutionResult
Source code in apogee_ai_sandbox/infrastructure/adapters/e2b_sandbox.py
Python
async def run(self, request: ExecutionRequest) -> ExecutionResult:
if not self.supports(request.language):
raise LanguageNotSupportedException(request.language.value, self.name)
try:
from e2b_code_interpreter import Sandbox # type: ignore
except ImportError as exc: # pragma: no cover
raise ImportError(str(exc)) from exc
job_id = secrets.token_hex(8)
start = time.perf_counter()
def _run() -> ExecutionResult:
sbx = Sandbox(template=self._template, api_key=self._api_key)
try:
# Materialise auxiliary files
for f in request.files:
sbx.files.write(f.path, f.content)
execution = sbx.run_code(request.code)
stdout = "".join(execution.logs.stdout)
stderr = "".join(execution.logs.stderr)
error = execution.error.value if getattr(execution, "error", None) else None
status = JobStatus.SUCCEEDED if not error else JobStatus.FAILED
return ExecutionResult(
job_id=job_id,
backend=self.name,
status=status,
stdout=stdout,
stderr=stderr,
error=error,
duration_ms=(time.perf_counter() - start) * 1000.0,
)
finally:
try:
sbx.kill()
except Exception: # noqa: BLE001
pass
try:
return await asyncio.wait_for(
asyncio.to_thread(_run), timeout=request.quotas.wall_seconds
)
except TimeoutError:
return ExecutionResult(
job_id=job_id,
backend=self.name,
status=JobStatus.TIMEOUT,
error=f"wall_seconds={request.quotas.wall_seconds} exceeded",
duration_ms=(time.perf_counter() - start) * 1000.0,
)
async
¶
Python
stream(request: ExecutionRequest) -> AsyncIterator[StreamChunk]
Source code in apogee_ai_sandbox/infrastructure/adapters/e2b_sandbox.py
Python
async def stream(self, request: ExecutionRequest) -> AsyncIterator[StreamChunk]:
result = await self.run(request)
async def gen() -> AsyncIterator[StreamChunk]:
if result.stdout:
yield StreamChunk(job_id=result.job_id, channel="stdout", payload=result.stdout)
yield StreamChunk(
job_id=result.job_id, channel="status", payload=result.status.value
)
return gen()
async
¶
async
¶
Toy sandbox that does not execute the code — just echoes it back.
Useful for tests, dry-runs, contract verification of plumbing without paying execution latency or risk.
async
¶
Python
run(request: ExecutionRequest) -> ExecutionResult
Source code in apogee_ai_sandbox/infrastructure/adapters/echo_sandbox.py
async
¶
Python
stream(request: ExecutionRequest) -> AsyncIterator[StreamChunk]
Source code in apogee_ai_sandbox/infrastructure/adapters/echo_sandbox.py
Python
async def stream(self, request: ExecutionRequest) -> AsyncIterator[StreamChunk]:
result = await self.run(request)
async def gen() -> AsyncIterator[StreamChunk]:
yield StreamChunk(job_id=result.job_id, channel="stdout", payload=result.stdout)
yield StreamChunk(job_id=result.job_id, channel="status", payload="succeeded")
return gen()
async
¶
async
¶
Best-effort static analysis gate.
Scans literals (and any explicit URLs) inside the code for hostnames that aren't on the allowlist. The gate is complementary to runtime isolation — it catches obvious leaks at the boundary so they never even reach the sandbox.
Source code in apogee_ai_sandbox/infrastructure/quota/egress_allowlist_gate.py
Python
check(*, code: str, allowlist: EgressAllowlist) -> None
Source code in apogee_ai_sandbox/infrastructure/quota/egress_allowlist_gate.py
Bases: _JailSandbox
Source code in apogee_ai_sandbox/infrastructure/adapters/jail_sandbox.py
Source code in apogee_ai_sandbox/infrastructure/cache/in_memory_fingerprint_cache.py
async
¶
Python
get(key: OutputCacheKey) -> ExecutionResult | None
Source code in apogee_ai_sandbox/infrastructure/cache/in_memory_fingerprint_cache.py
Python
async def get(self, key: OutputCacheKey) -> ExecutionResult | None:
cached = self._store.get((key.fingerprint.digest, key.backend))
if cached is None:
return None
# Mark as cached on the way out without mutating the stored copy
return ExecutionResult(
job_id=cached.job_id,
backend=cached.backend,
status=cached.status,
stdout=cached.stdout,
stderr=cached.stderr,
exit_code=cached.exit_code,
duration_ms=cached.duration_ms,
cpu_seconds=cached.cpu_seconds,
memory_peak_mb=cached.memory_peak_mb,
artifacts=dict(cached.artifacts),
error=cached.error,
cached=True,
)
async
¶
Python
put(key: OutputCacheKey, result: ExecutionResult) -> None
async
¶
Python
InMemoryQuotaEnforcer(*, cpu_seconds_per_minute: float = 600.0, max_inflight: int = 8, raise_on_exceeded: bool = True)
Sliding-window quota gate per tenant.
Default budget: 600 CPU-seconds per 60s window, 8 concurrent jobs.
Pass raise_on_exceeded=True to surface QuotaExceededException;
otherwise acquire returns False.
Source code in apogee_ai_sandbox/infrastructure/quota/in_memory_quota_enforcer.py
Python
def __init__(
self,
*,
cpu_seconds_per_minute: float = 600.0,
max_inflight: int = 8,
raise_on_exceeded: bool = True,
) -> None:
self._budget = cpu_seconds_per_minute
self._max_inflight = max_inflight
self._raise = raise_on_exceeded
self._buckets: dict[str, _Bucket] = defaultdict(
lambda: _Bucket(self._budget, max_inflight=self._max_inflight)
)
self._lock = asyncio.Lock()
async
¶
Python
acquire(*, tenant_id: str | None, quotas: Quotas) -> bool
Source code in apogee_ai_sandbox/infrastructure/quota/in_memory_quota_enforcer.py
Python
async def acquire(self, *, tenant_id: str | None, quotas: Quotas) -> bool:
key = tenant_id or "_default"
async with self._lock:
bucket = self._buckets[key]
self._refresh(bucket)
if bucket.inflight >= bucket.max_inflight:
if self._raise:
raise QuotaExceededException(
"concurrent_jobs", bucket.max_inflight, bucket.inflight + 1
)
return False
projected = bucket.cpu_seconds_used + quotas.cpu_seconds
if projected > bucket.cpu_seconds_per_minute:
if self._raise:
raise QuotaExceededException(
"cpu_seconds_per_minute",
bucket.cpu_seconds_per_minute,
projected,
)
return False
bucket.inflight += 1
return True
async
¶
Source code in apogee_ai_sandbox/infrastructure/quota/in_memory_quota_enforcer.py
Python
async def release(self, *, tenant_id: str | None, observed_cpu_seconds: float) -> None:
key = tenant_id or "_default"
async with self._lock:
bucket = self._buckets[key]
bucket.cpu_seconds_used = max(
0.0, bucket.cpu_seconds_used + max(0.0, observed_cpu_seconds)
)
bucket.inflight = max(0, bucket.inflight - 1)
async
¶
Source code in apogee_ai_sandbox/infrastructure/quota/in_memory_quota_enforcer.py
Python
async def remaining(self, *, tenant_id: str | None) -> dict[str, float]:
key = tenant_id or "_default"
async with self._lock:
bucket = self._buckets[key]
self._refresh(bucket)
return {
"cpu_seconds_remaining": max(
0.0, bucket.cpu_seconds_per_minute - bucket.cpu_seconds_used
),
"inflight": float(bucket.inflight),
"max_inflight": float(bucket.max_inflight),
}
File-based cache: <root>/<backend>/<digest>.json.
Source code in apogee_ai_sandbox/infrastructure/cache/json_fingerprint_cache.py
async
¶
Python
get(key: OutputCacheKey) -> ExecutionResult | None
async
¶
Python
put(key: OutputCacheKey, result: ExecutionResult) -> None
async
¶
Subprocess-based runner — not a real sandbox.
Limits CPU/memory via resource.RLIMIT_* (POSIX only) and isolates
workspace to a temporary directory. Useful in dev, but not safe
against malicious code on the host. Production must prefer Docker / E2B
/ Modal / Firejail.
Network policy NONE is enforced via env-var hints (NO_PROXY,
HTTP_PROXY=http://0.0.0.0:1) — best-effort only.
Source code in apogee_ai_sandbox/infrastructure/adapters/local_process_sandbox.py
Python
def __init__(self, *, allow_unsafe: bool = False) -> None:
if not allow_unsafe and not self._is_dev_environment():
# Default: refuse to run with the local backend in production-like
# environments (CI=true) without the explicit override.
self._unsafe_disabled = True
else:
self._unsafe_disabled = False
self._jobs: dict[str, asyncio.subprocess.Process] = {}
async
¶
Python
run(request: ExecutionRequest) -> ExecutionResult
Source code in apogee_ai_sandbox/infrastructure/adapters/local_process_sandbox.py
Python
async def run(self, request: ExecutionRequest) -> ExecutionResult:
if self._unsafe_disabled:
return ExecutionResult(
job_id=secrets.token_hex(8),
backend=self.name,
status=JobStatus.FAILED,
error=(
"LocalProcessSandbox is disabled outside dev. "
"Pass allow_unsafe=True if you understand the risks."
),
)
if not self.supports(request.language):
raise LanguageNotSupportedException(request.language.value, self.name)
if request.egress.policy == NetworkPolicy.UNRESTRICTED and not self._is_dev_environment():
raise EgressBlockedException("unrestricted-not-allowed")
job_id = secrets.token_hex(8)
with tempfile.TemporaryDirectory(prefix="apogee-sandbox-") as workdir_str:
workdir = Path(workdir_str)
self._materialize_files(workdir, request)
cmd = self._build_command(workdir, request)
env = self._build_env(request)
start = time.perf_counter()
try:
proc = await asyncio.create_subprocess_exec(
*cmd,
stdout=asyncio.subprocess.PIPE,
stderr=asyncio.subprocess.PIPE,
cwd=str(workdir),
env=env,
preexec_fn=self._build_preexec(request),
)
except FileNotFoundError as exc:
return ExecutionResult(
job_id=job_id,
backend=self.name,
status=JobStatus.FAILED,
error=f"interpreter not found: {exc}",
)
self._jobs[job_id] = proc
try:
stdout, stderr = await asyncio.wait_for(
proc.communicate(), timeout=request.quotas.wall_seconds
)
duration_ms = (time.perf_counter() - start) * 1000.0
exit_code = proc.returncode
status = JobStatus.SUCCEEDED if exit_code == 0 else JobStatus.FAILED
return ExecutionResult(
job_id=job_id,
backend=self.name,
status=status,
stdout=stdout.decode(errors="replace"),
stderr=stderr.decode(errors="replace"),
exit_code=exit_code,
duration_ms=duration_ms,
)
except TimeoutError:
proc.kill()
await proc.wait()
return ExecutionResult(
job_id=job_id,
backend=self.name,
status=JobStatus.TIMEOUT,
error=f"wall_seconds={request.quotas.wall_seconds} exceeded",
duration_ms=(time.perf_counter() - start) * 1000.0,
)
finally:
self._jobs.pop(job_id, None)
async
¶
Python
stream(request: ExecutionRequest) -> AsyncIterator[StreamChunk]
Source code in apogee_ai_sandbox/infrastructure/adapters/local_process_sandbox.py
Python
async def stream(self, request: ExecutionRequest) -> AsyncIterator[StreamChunk]:
result = await self.run(request)
async def gen() -> AsyncIterator[StreamChunk]:
if result.stdout:
yield StreamChunk(job_id=result.job_id, channel="stdout", payload=result.stdout)
if result.stderr:
yield StreamChunk(job_id=result.job_id, channel="stderr", payload=result.stderr)
yield StreamChunk(
job_id=result.job_id, channel="status", payload=result.status.value
)
return gen()
async
¶
async
¶
Adapter for modal.com Function-based sandboxes.
Lazy import: install via pip install 'apogee-ai-sandbox[modal]'.
Source code in apogee_ai_sandbox/infrastructure/adapters/modal_sandbox.py
async
¶
Python
run(request: ExecutionRequest) -> ExecutionResult
Source code in apogee_ai_sandbox/infrastructure/adapters/modal_sandbox.py
Python
async def run(self, request: ExecutionRequest) -> ExecutionResult:
if not self.supports(request.language):
raise LanguageNotSupportedException(request.language.value, self.name)
try:
import modal # type: ignore
except ImportError as exc: # pragma: no cover
raise ImportError(str(exc)) from exc
job_id = secrets.token_hex(8)
start = time.perf_counter()
def _run() -> ExecutionResult:
app = modal.App.lookup(self._app_name, create_if_missing=True)
sandbox = modal.Sandbox.create(
app=app,
image=modal.Image.debian_slim().pip_install([]),
cpu=request.quotas.cpu_seconds,
memory=request.quotas.memory_mb,
timeout=int(request.quotas.wall_seconds),
block_network=request.egress.policy.value == "none",
)
try:
if request.language == Language.PYTHON:
proc = sandbox.exec("python", "-c", request.code)
else:
proc = sandbox.exec("sh", "-c", request.code)
stdout = proc.stdout.read()
stderr = proc.stderr.read()
exit_code = proc.wait()
status = JobStatus.SUCCEEDED if exit_code == 0 else JobStatus.FAILED
return ExecutionResult(
job_id=job_id,
backend=self.name,
status=status,
stdout=stdout,
stderr=stderr,
exit_code=exit_code,
duration_ms=(time.perf_counter() - start) * 1000.0,
)
finally:
try:
sandbox.terminate()
except Exception: # noqa: BLE001
pass
try:
return await asyncio.wait_for(
asyncio.to_thread(_run),
timeout=request.quotas.wall_seconds + 30,
)
except TimeoutError:
return ExecutionResult(
job_id=job_id,
backend=self.name,
status=JobStatus.TIMEOUT,
error=f"wall_seconds={request.quotas.wall_seconds} exceeded",
duration_ms=(time.perf_counter() - start) * 1000.0,
)
async
¶
Python
stream(request: ExecutionRequest) -> AsyncIterator[StreamChunk]
Source code in apogee_ai_sandbox/infrastructure/adapters/modal_sandbox.py
Python
async def stream(self, request: ExecutionRequest) -> AsyncIterator[StreamChunk]:
result = await self.run(request)
async def gen() -> AsyncIterator[StreamChunk]:
if result.stdout:
yield StreamChunk(job_id=result.job_id, channel="stdout", payload=result.stdout)
yield StreamChunk(
job_id=result.job_id, channel="status", payload=result.status.value
)
return gen()
async
¶
async
¶
Python
NotebookRunnerTool(sandbox: ISandbox, *, default_quotas: Quotas | None = None, default_egress: EgressAllowlist | None = None)
Bases: SandboxTool
Source code in apogee_ai_sandbox/infrastructure/tools/sandbox_tools.py
Python
SandboxRegistry(sandboxes: Mapping[str, ISandbox] | None = None)
In-memory map backend_name → ISandbox.
The registry deliberately avoids dynamic imports — callers pass already instantiated sandboxes (with the right extras already installed).
Source code in apogee_ai_sandbox/infrastructure/registry/sandbox_registry.py
Python
SandboxTool(sandbox: ISandbox, *, default_quotas: Quotas | None = None, default_egress: EgressAllowlist | None = None)
Base class wrapping an :class:ISandbox as a typed agent tool.
Source code in apogee_ai_sandbox/infrastructure/tools/sandbox_tools.py
Python
definition() -> SandboxToolDefinition
async
¶
Python
execute(*, code: str, tenant_id: str | None = None, user_id: str | None = None, files: tuple[SandboxFile, ...] | None = None) -> ExecutionResult
Source code in apogee_ai_sandbox/infrastructure/tools/sandbox_tools.py
Python
async def execute(
self,
*,
code: str,
tenant_id: str | None = None,
user_id: str | None = None,
files: tuple[SandboxFile, ...] | None = None,
) -> ExecutionResult:
request = ExecutionRequest(
code=code,
language=self.language,
files=files or (),
quotas=self._default_quotas,
egress=self._default_egress,
tenant_id=tenant_id,
user_id=user_id,
)
return await self._sandbox.run(request)
dataclass
¶
Python
ShellRunnerTool(sandbox: ISandbox, *, default_quotas: Quotas | None = None, default_egress: EgressAllowlist | None = None)
Bases: SandboxTool
Source code in apogee_ai_sandbox/infrastructure/tools/sandbox_tools.py
Python
SqlRunnerTool(sandbox: ISandbox, *, default_quotas: Quotas | None = None, default_egress: EgressAllowlist | None = None)
Bases: SandboxTool
Wraps ISandbox to run SQL by piping code through psql/mysql.
The tool itself is sandbox-agnostic — the actual SQL client must be
available in the sandbox image. Useful with DockerSandboxAdapter and
a custom image that already contains psql.
Source code in apogee_ai_sandbox/infrastructure/tools/sandbox_tools.py
class-attribute
instance-attribute
¶
Python
description = 'Execute a SQL script (engine-agnostic; sandbox image must include the client).'
async
¶
Python
execute(*, code: str, tenant_id: str | None = None, user_id: str | None = None, files: tuple[SandboxFile, ...] | None = None, engine: str = 'psql', connection: str = '') -> ExecutionResult
Source code in apogee_ai_sandbox/infrastructure/tools/sandbox_tools.py
Python
async def execute( # type: ignore[override]
self,
*,
code: str,
tenant_id: str | None = None,
user_id: str | None = None,
files: tuple[SandboxFile, ...] | None = None,
engine: str = "psql",
connection: str = "",
) -> ExecutionResult:
wrapper = (
f'echo {json.dumps(code)} | {engine} {connection}'
if engine != "echo"
else code
)
request = ExecutionRequest(
code=wrapper,
language=Language.BASH,
files=files or (),
quotas=self._default_quotas,
egress=self._default_egress,
tenant_id=tenant_id,
user_id=user_id,
)
return await self._sandbox.run(request)
Python
default_tool_catalog(sandbox: ISandbox) -> dict[str, SandboxTool]
Return the four built-in tools wired to sandbox.
Source code in apogee_ai_sandbox/infrastructure/tools/sandbox_tools.py
Python
def default_tool_catalog(sandbox: ISandbox) -> dict[str, SandboxTool]:
"""Return the four built-in tools wired to ``sandbox``."""
return {
CodeInterpreterTool.name: CodeInterpreterTool(sandbox),
ShellRunnerTool.name: ShellRunnerTool(sandbox),
NotebookRunnerTool.name: NotebookRunnerTool(sandbox),
SqlRunnerTool.name: SqlRunnerTool(sandbox),
}