Python SDK

faber-sdk is the Python client for the Faber API. It has a synchronous Client and an asynchronous AsyncClient with the same methods, typed models for every schema in the API contract, a fluent builder for steps, and automatic retries for transient failures. It needs Python 3.10 or newer; its only dependencies are httpx and anyio (which httpx uses anyway).

The source lives in sdks/python.

Install

pip install faber-sdk
# or
uv add faber-sdk

The distribution is called faber-sdk; the package you import is faber. It ships py.typed, so mypy and pyright see every type.

Configure

from faber import Client

faber = Client("http://localhost:3000", api_key="your-api-key")

With no arguments, Client() reads FABER_URL and FABER_API_KEY from the environment and falls back to http://localhost:3000. The base URL may be the service root (http://host:3000) or the API root (http://host:3000/api/v1); both work, and so does a path prefix such as https://example.com/faber.

Argument	Default	Meaning
base_url	$FABER_URL, else http://localhost:3000	Service root or API root
api_key	$FABER_API_KEY	Sent as Authorization: Bearer <key>
timeout	60 seconds	A number of seconds or an httpx.Timeout
max_retries	2	Retries after the first attempt (503, 429 and connection errors)
retry_backoff	0.5 seconds	First retry delay; doubles each retry, with jitter, capped at 60 s
headers	none	Extra headers sent with every request
http_client	a new httpx.Client	Your own client, for proxies, TLS settings or custom transports

Use the client as a context manager, or call close(), to release its connections. A client you pass in through http_client is never closed by the SDK, and it keeps its own timeout unless you also pass timeout.

import httpx
from faber import Client

http = httpx.Client(proxy="http://proxy.internal:8080", verify="/etc/ssl/internal-ca.pem")
with Client("https://faber.internal", api_key="...", http_client=http, timeout=120) as faber:
    print(faber.health())

Retries

The client retries a request when the server answers 503 (no free execution slots, or the service is draining) or 429 (the API key is rate limited), and when the connection fails before a response arrives. On a 503 or 429 it waits for the Retry-After header if the server sends one (capped at 60 seconds), otherwise it backs off exponentially. It never retries other 4xx responses or other 5xx responses, and it does not retry a request that timed out while waiting for the response, because the server may still be running it. Uploads from a non-seekable stream are not retried either, since the body cannot be sent twice. Set max_retries=0 to turn retries off.

Run code

A request is a list of steps that run in order in one sandbox workspace (/faber). A step is a single task, or a list of tasks that run in parallel. Build steps with task() and StepBuilder, then pass them to execute:

from faber import Client, StepBuilder, task

source = '#include <stdio.h>\nint main(void){int a,b;scanf("%d %d",&a,&b);printf("%d\\n",a+b);}\n'

steps = (
    StepBuilder()
    .step(task("gcc", "-O2", "-o", "main", "main.c").file("main.c", source).cache())
    .parallel(
        task("./main").stdin("1 2\n").profile("native_v2"),
        task("./main").stdin("40 2\n").profile("native_v2"),
    )
)

with Client() as faber:
    response = faber.execute(steps, stop_on_failure=True)

print(response.ok, response.cache.hits)

execute accepts:

• a StepBuilder,
• a list of steps, where each step is a Task, a TaskBuilder, or a list of them for a parallel step: faber.execute([task("make"), [task("./a"), task("./b")]]),
• an ExecuteRequest model.

The keyword arguments set the request envelope and override the values of an ExecuteRequest you pass:

Argument	Meaning
cache	Default for tasks that do not set cache themselves. Only a leading run of cacheable steps is ever cached. Ignored when the server cache is disabled.
stop_on_failure	After a step in which any task did not exit with status 0, report the remaining steps as skipped (not_started) instead of running them.

To run a single task and get its result directly, use run:

from faber import CompletedTaskResult, task

result = faber.run(task("echo", "hello"))
if isinstance(result, CompletedTaskResult):
    print(result.stdout)

Models instead of the builder

The builder produces plain frozen dataclasses, which you can also create directly:

from faber import ExecuteRequest, FileReference, InlineFile, SandboxProfile, Task

request = ExecuteRequest(
    [
        Task(
            "gcc",
            args=["-O2", "-o", "main", "main.c"],
            cache=True,
            files={
                "main.c": source,  # a plain string is UTF-8 content
                "data/blob.bin": InlineFile.binary(b"\x00\x01\x02"),
                "run.sh": InlineFile("#!/bin/sh\nexec ./main\n", executable=True),
                "input.txt": FileReference("9f86d081884c7d65..."),
            },
        ),
        [
            Task("./main", stdin="1 2\n", sandbox_profile=SandboxProfile.NATIVE_V2),
            Task("./main", stdin="40 2\n", sandbox_profile=SandboxProfile.NATIVE_V2),
        ],
    ],
    stop_on_failure=True,
)
response = faber.execute(request)

Every model has from_dict() and to_dict() to convert from and to the JSON form of the API.

The step builder

task(cmd, *args) starts a TaskBuilder. Each method sets one field and returns the builder; build() returns the Task. Builders are accepted anywhere a task is, so you rarely call build() yourself.

Method	Effect
.args(*args)	Append command-line arguments
.env(mapping, **vars)	Set environment variables
.stdin(text)	Text written to standard input, which is then closed
.file(path, content, executable=False)	Create a workspace file from str (sent as text) or bytes (sent base64-encoded)
.file_ref(path, file, executable=False)	Create a workspace file from an uploaded file: its id, an UploadResponse or a FileInfo
.files(mapping)	Add several files (str, InlineFile or FileReference values)
.working_dir(path)	Absolute directory inside the sandbox to start in (default /faber)
.profile(profile)	Sandbox profile, SandboxProfile or string, e.g. "native_v2"
.cache(enabled=True)	Declare the task deterministic so its step may be cached; cache(False) opts out

File paths are relative to the workspace, without .. or a leading /.

StepBuilder collects the steps:

Method	Effect
.step(task)	Add a step that runs one task
.parallel(*tasks)	Add a step whose tasks run concurrently, each in its own PID namespace and cgroup
.build()	The list of steps
.request(cache=None, stop_on_failure=None)	An ExecuteRequest with these steps

Read results

ExecuteResponse.steps mirrors the request: step i of the response is a single TaskResult when step i was a task, and a list of them (in the same order) when it was a parallel step. ExecuteResponse.cache reports whether the server cache is enabled, how many steps were replayed (hits) and how many were stored.

A TaskResult is one of two dataclasses:

• CompletedTaskResult: the process ran. Fields stdout, stderr, exit_code (128 + signal number when killed by a signal), stats and cached.
• FailedTaskResult: the task never reached an exit status (setup failure, command not found, or skipped by stop_on_failure). Fields error, stats and cached.

Both have ok (exited with status 0) and outcome (shortcut for stats.outcome). Narrow with isinstance or match:

from faber import CompletedTaskResult, FailedTaskResult

for result in response.task_results():
    match result:
        case CompletedTaskResult(exit_code=0, stdout=out):
            print("ok:", out.strip())
        case CompletedTaskResult(exit_code=code, outcome=outcome):
            print("failed:", code, outcome)
        case FailedTaskResult(error=error):
            print("did not run:", error)

Helpers on the response:

Helper	Meaning
response.ok	Every task exited with status 0
response.task_results()	Iterate every task result in request order, parallel steps flattened
response.first_failure()	The first result that did not exit with status 0, or None

TaskStats holds memory_peak_bytes, cpu_usage_usec, pids_peak, execution_time_ms, stdout_truncated, stderr_truncated, outcome, termination_signal and the optional cgroup counters. cached results carry the stats of the run that was stored. The outcome tells you how the task ended:

TaskOutcome	Meaning
EXITED	The process exited on its own; check exit_code
SIGNALED	Ended by a signal that none of the limits below explains
TIMED_OUT	Exceeded the wall-clock or CPU time limit
OUT_OF_MEMORY / ANCESTOR_OUT_OF_MEMORY	Killed by the cgroup memory limit
PIDS_LIMIT	Hit the process count limit
OUTPUT_LIMIT	Wrote more output than allowed
POLICY_VIOLATION	Made a system call its sandbox profile denies (SIGSYS)
NOT_STARTED	Skipped because an earlier step failed
INFRASTRUCTURE_FAILURE	The sandbox could not run the task

Files

Uploaded files are stored under the SHA-256 of their content, so uploading the same bytes twice returns the same id with already_exists=True. Reference a stored file from any task instead of sending its content again.

uploaded = faber.upload_file(b"42\n7\n19\n", filename="numbers.txt", content_type="text/plain")
faber.upload_file("data/large.csv")            # a path: the file name is sent too
with open("model.bin", "rb") as f:
    faber.upload_file(f)                       # a binary file object

result = faber.run(task("sort", "-n", "numbers.txt").file_ref("numbers.txt", uploaded))

for info in faber.list_files():                # list[FileInfo]
    print(info.id, info.size, info.filename, info.created_at)

data = faber.download_file(uploaded)           # bytes
faber.delete_file(uploaded)

The file methods take a file id, an UploadResponse or a FileInfo. Referencing an id that is not stored makes execute fail with 404 and the code file_not_found.

Service operations

Method	Route	Returns
health()	GET /health	HealthResponse with status and version. A draining service returns status == HealthStatus.SHUTTING_DOWN instead of raising. No API key needed.
info()	GET /info	ServiceInfo: version, architecture, allowed and default sandbox profiles, every effective limit, cache settings and installed toolchains (info.toolchain("python"))
clear_cache()	DELETE /cache	CacheClearResponse with the number of layers removed. Raises cache_disabled when the cache is off.
metrics()	GET /metrics	The Prometheus text exposition, as str

Errors

Every failed request raises faber.FaberError:

Attribute	Meaning
status	HTTP status, or None when no response arrived (connection refused, timeout)
code	ErrorCode from the error body, or None when the body had none
message	The server’s message
request_id	The x-request-id response header; quote it when reading server logs

For transport failures the original httpx exception is the __cause__.

from faber import ErrorCode, FaberError

try:
    response = faber.execute(steps)
except FaberError as err:
    if err.code in (ErrorCode.CAPACITY_EXHAUSTED, ErrorCode.RATE_LIMITED):
        ...  # still busy or rate limited after the automatic retries
    elif err.code is ErrorCode.PROFILE_NOT_ALLOWED:
        ...  # the service does not allow the requested sandbox profile
    else:
        raise

Status	Codes
400	invalid_request
401	unauthorized
404	not_found, file_not_found
409	cache_disabled
413	payload_too_large
422	limits_exceeded, profile_not_allowed
429	rate_limited (retried automatically)
500	sandbox_setup_failed, internal_error
503	capacity_exhausted, shutting_down, cancelled (retried automatically)
504	deadline_exceeded

A task that fails is not an error. Non-zero exits, timeouts, OOM kills and policy violations come back in the task’s result with 200 OK.

Async

AsyncClient has the same constructor and methods, as coroutines. It works under asyncio and trio.

import asyncio
from faber import AsyncClient, task

async def main() -> None:
    async with AsyncClient() as faber:
        results = await asyncio.gather(*(faber.run(task("echo", str(i))) for i in range(5)))
        print([r.ok for r in results])

asyncio.run(main())

Pass http_client=httpx.AsyncClient(...) to reuse your own client.

Forward compatibility

Enums such as TaskOutcome, ErrorCode and SandboxProfile accept values they do not list. If a newer server sends a new outcome, it decodes to a member that keeps the raw value, compares equal to the string, and has is_known == False; the rest of the response decodes normally. Unknown JSON fields are ignored.

Examples

The examples/ directory has runnable scripts. Each reads FABER_URL and FABER_API_KEY:

cd sdks/python
export FABER_URL=http://localhost:3000 FABER_API_KEY=your-api-key
uv run examples/hello_world.py        # run one command
uv run examples/compile_c.py          # compile C, run it with input on stdin
uv run examples/cached_test_cases.py  # compile once (cached), run test cases in parallel, twice
uv run examples/upload_file.py        # upload a file and use it as a task input
uv run examples/errors_and_limits.py  # FaberError vs timeout, OOM and policy violation outcomes

The C examples need gcc in the Faber image (the default vgfractal/faber image has it). The cache example shows hits on its second run only when the server runs with CACHE_ENABLED=true.