The flamepy package provides a synchronous client for Flame sessions and tasks, a service base class for host-shim applications, object-cache helpers, the App API for packaging Python workloads, and the Agent Session API for remote script execution.
By default the SDK reads ~/.flame/flame.yaml:
---
current-context: flame
contexts:
- name: flame
cluster:
endpoint: "http://127.0.0.1:8080"
cache:
endpoint: "grpc://127.0.0.1:9090"Environment variables override the file:
FLAME_ENDPOINTFLAME_CACHE_ENDPOINTFLAME_CACHE_STORAGEFLAME_CA_FILE
Top-level helpers use the configured default connection:
import flamepy
session = flamepy.create_session(
"flmping",
min_instances=1,
resreq=flamepy.ResourceRequirement.from_string("cpu=1,mem=1g"),
)
output = session.run(b"hello")
session.close()Public helpers:
connect(addr, tls_config=None) -> Connectioncreate_session(application, common_data=None, session_id=None, min_instances=0, max_instances=None, batch_size=1, resreq=None) -> Sessionopen_session(session_id, spec=None) -> Sessionregister_application(name, app_attrs) -> Noneunregister_application(name) -> Nonelist_applications() -> list[Application]get_application(name) -> Application | Nonelist_sessions() -> list[Session]get_session(session_id) -> Sessionclose_session(session_id) -> Session
Session represents an open or closed Flame session.
Methods:
create_task(input_data: bytes) -> Taskget_task(task_id) -> Tasklist_tasks() -> Iterator[Task]watch_task(task_id, timeout=None) -> TaskWatcherrun(input_data) -> bytessubmit(input_data) -> TaskFutureclose() -> Nonecommon_data() -> bytes | None
create_task() expects bytes. run() blocks until the task completes and
returns its output. submit() returns immediately with a TaskFuture, a
concurrent.futures.Future subclass that resolves with the task output and
dispatches user completion callbacks away from the shared aio loop.
SessionAttributes:
application: strid: str | Nonecommon_data: bytes | Nonemin_instances: int(long-lived instances to keep; not a gang)max_instances: int | Nonebatch_size: int(gang size: executors are bound only once this many can be bound together;1is not a gang)resreq: ResourceRequirement | None
ApplicationAttributes:
shim: Shim | Noneimage: str | Nonedescription: str | Nonelabels: list[str] | Nonecommand: str | Nonearguments: list[str] | Noneenvironments: dict[str, str] | Noneworking_directory: str | Nonemax_instances: int | Nonedelay_release: int | Noneschema: ApplicationSchema | Noneurl: str | Noneinstaller: str | None
ResourceRequirement.from_string("cpu=1,mem=1g,gpu=0") parses user-friendly resource strings into CPU, memory bytes, and GPU counts.
Top-level flamepy exports:
ObjectRefput_object(key_prefix, obj)get_object(ref, deserializer=None)
flamepy.core also exports:
ObjectKeyWILDCARD_SESSIONpatch_object(ref, delta)upload_package(app_name, file_path, filename=None)download_package(app_name, filename, dest_path)delete_package(app_name, filename)
flamepy.cache exports delete_objects(key_prefix) in addition to the basic cache helpers.
Object references are versioned:
version=0forces a fresh download.version>=1lets the client reuse its cached parts and fetch only the data after that version.- Without a deserializer,
get_object()returns the newest data part of the object. - With a deserializer,
get_object(ref, deserializer)callsdeserializer(parts)with all the data parts in the order they were written.
Host-shim Python services subclass flamepy.FlameService and run with flamepy.run():
import flamepy
class Echo(flamepy.FlameService):
def on_session_enter(self, context):
self.session_id = context.session_id
def on_task_invoke(self, context):
return context.input
def on_session_leave(self):
pass
if __name__ == "__main__":
flamepy.run(Echo())Service contexts expose bytes-oriented APIs:
SessionContext.session_idSessionContext.applicationSessionContext.common_data()TaskContext.task_idTaskContext.session_idTaskContext.inputFlameService.publish(keys)andretract(keys)publish and withdraw data-aware scheduling keys for the next response; aflamepy.serving.Instancehas them too.- On the client,
flamepy.serving.Session.run(req, option=None)andsubmit(req, option=None)acceptTaskOptions(affinity={key})to prefer an executor that publishedkey.
The process-wide application API is exported from flamepy.app:
import flamepy.app as app
app.init("add-app")
@app.service()
def add(a, b):
return a + b
print(add(1, 2).get())
app.destroy()A service may invoke another declared service. The captured proxy is restored in the executor as a reference to its existing session:
@app.service()
def fn_a(value):
return value * 2
@app.service()
def fn_b(value):
return fn_a(value).get()Key classes and helpers:
app.session_context()returns the active service-side session context (flamepy.SessionContext) during an App invocation.app.context()returns anapp.InstanceContextwith thesession_id,task_idandbatch_sizeof the current App invocation.batch_sizeis the session's gang size: while the session has pending tasks, the scheduler binds its executors all at once or not at all. A task that blocks untilbatch_sizemembers have joined, such as a collectiveinit_process_group, can use it as the world size. It raisesRuntimeErroroutside an invocation.app.publish_attributes(attrs)adds opaquebyteslocality keys to the current App response; repeated calls in the response accumulate.app.retract_attributes(attrs)retracts keys that are no longer valid, such as an evicted cache entry. The last call on a key within a response wins. Session Manager adds published keys to the executor's retained set and removes retracted ones; retracting an absent key is harmless.init(name, fail_if_exists=False, dependencies=None, python_version=None)packages and initializes the application and returns its runtime handle. Repeating it for the same name returns the same handle.dependenciespopulates an auto-generatedpyproject.tomlonly when the project has no existing Python package metadata.python_versionselects the executor Python version through the application template.app.service(autoscale=None, warmup=0, resreq=None, batch_size=1)is the canonical service decorator. Functions becomeServiceInstanceproxies. Decorating a class declares a service class without creating a session. CallingDecoratedClass(args...)creates its service handle and session, andflmrunruns the constructor with those arguments in the executor. The decorator accepts functions and classes, not already constructed objects.- Decorated classes do not expose class-level service methods. Invoke methods
directly on a constructed handle, for example
instance.method(...); Flame does not require a.remote()suffix. - Each constructed handle has a unique service ID and its own retained object;
two handles created from the same decorated class do not share object state.
The retained object is executor-local: autoscaling may create one copy per
executor. Use
autoscale=False, warmup=1when mutable state must stay on one executor. warmupkeeps that many instances alive (min_instances); it does not make them a gang.batch_sizeis the gang size: usebatch_size=Nfor work that blocks until N members are present, together withwarmup=Nto keep them. For a gangwarmupmust be 0 or N, andautoscale=Falsewithwarmup=Nalso fixes the maximum at N; any othermin_instancesormax_instancesis rejected when the session is created.- Public class methods must not collide with
ServiceInstanceAPI names such asclose; handle construction rejects such definitions. - The handle returned by
app.init()also exposesservice(...)when direct runtime access is useful. A declaration made during another App invocation is the sole no-init exception; it reuses the current session. ServiceInstanceis the client proxy returned for a decorated function or by calling a decorated class with its remote constructor arguments.- A recursive declaration may call the service decorator without
init()and does not calldestroy(); it reuses and does not own the parent session. Nested submission works at any capacity. If the parent waits synchronously, the session needs enough free executor capacity to run the nested task. get(futures),ref(futures),wait(futures),select(futures)put(obj)stores a shared object under the active application's cache prefix.destroy()closes services. If this process registered the application, it also unregisters it and removes its package/cache. Withfail_if_exists=False, an existing application is borrowed and remains the user's lifecycle responsibility.ObjectFuture.get(),ObjectFuture.ref(),ObjectFuture.wait()
The agent Session lives under flamepy.agent. It runs remote Python or shell scripts in sandboxed containers through the built-in flmbox application, the same backend as Sandbox, without exposing the core Session or task JSON.
from flamepy.agent import open_session
from flamepy import ResourceRequirement, FlameError
with open_session() as ssn:
print(ssn.run_code("echo hello").stdout_text())Key classes and methods:
open_session(*, ssn_id=None, image=None, min_instances=0, max_instances=None, resreq=None)open_session()creates a session in theflmboxservice's default image; passimageto choose the sandbox container image. To run Python, the image must provide it (for examplepython:3.12-slim)open_session(ssn_id="...")reopens the session with that ID, or creates it with the given options when it does not existSession.run_code(code, *, language="shell", cwd=None, envs=None, input=None, timeout=None) -> CommandResultblocks for the result;languageis"shell"(default) or"python"Session.submit_code(code, *, language="shell", cwd=None, envs=None, input=None, timeout=None) -> Future[CommandResult]returns immediatelySession.close()ssn.id,ssn.attrCommandResult.exit_code,.stdout,.stderr,.ok,.stdout_text(),.stderr_text(). A nonzero exit is a normal result, not an exception.
Creation options apply only when the session is created: an existing session keeps its own, so they are ignored when ssn_id names one that exists. ssn.attr is a read-only view of the effective creation settings; its type is intentionally not exported. image is create-time only. close() destroys the session. Reopen works only while the session is still open. ResourceRequirement and FlameError are imported from flamepy.
flamepy.agent.Sandbox runs commands and scripts inside a container sandbox served by the built-in flmbox application.
from flamepy.agent import Sandbox
with Sandbox.create(image="ubuntu:26.04") as box:
result = box.run("echo hello", cwd="/tmp", envs={"A": "1"}, timeout=30)
print(result.exit_code, result.stdout_text())
print(box.run_code("print(1 + 2)", language="python").stdout_text())Sandbox.create(*, sandbox_id=None, image=None, resreq=None)creates a sandbox (one instance). A missingsandbox_idgenerates asandbox-<random>id; an id that already exists is an error (useSandbox.connect). A missingimageuses the service default.resreqis the session resource requirement only; it is not part of the sandbox configuration.Sandbox.connect(sandbox_id)attaches to an existing sandbox; a session of another application raisesINVALID_ARGUMENT.box.id,box.state(a fresh query returning"open","suspending","suspended"or"closed")box.run(cmd, *, cwd=None, envs=None, timeout=None) -> CommandResultandbox.submit(...) -> Future[CommandResult]box.run_code(code, *, language="python", cwd=None, envs=None, input=None, timeout=None) -> CommandResultandbox.submit_code(...);languageis"python"or"shell",inputis bytes for stdin.box.suspend(*, timeout=60.0)requests suspension and polls until the state issuspended. The signal is asynchronous;INTERNALis raised on timeout.box.resume()returns when the sandbox isopenagain. Phase 1 does NOT preserve process or filesystem state: a resumed sandbox starts fresh.box.close()is idempotent; the context manager closes on exit.- A command is sent without asking the server for the state first. The server queues a command for a suspending or suspended sandbox and runs it after the resume, so
runblocks (andsubmitstays pending) until then; checkbox.statefirst if that matters. CommandResult.exit_code,.stdout,.stderr(bytes),.stdout_text(),.stderr_text(),.ok. A nonzero exit code is a normal result, not an error.
Errors are FlameError: INVALID_ARGUMENT for bad arguments, INVALID_STATE when this handle is closed (after close()); a command for a sandbox closed through another handle is rejected by the server with a FlameError saying the session is closed, and INTERNAL for a malformed service response or a suspend timeout.
SessionState.OPEN,SessionState.CLOSEDTaskState.PENDING,TaskState.RUNNING,TaskState.SUCCEED,TaskState.FAILEDApplicationState.ENABLED,ApplicationState.DISABLEDShim.HOST,Shim.WASMFlameErrorCode.INVALID_CONFIG,INVALID_STATE,INVALID_ARGUMENT,INTERNAL,ALREADY_EXISTS,NOT_FOUND
SDK operations raise FlameError with a code and message:
import flamepy
try:
flamepy.connect("invalid://address")
except flamepy.FlameError as exc:
print(exc.code, exc.message)