fal.api package

Submodules

fal.api.api module

class fal.api.api.BaseServable

Bases: object

collect_routes()

Return type:

dict[RouteSignature, Callable[..., Any]]

handle_exit()

lifespan(app)

openapi()

Build the OpenAPI specification for the served function. Attach needed metadata for a better integration to fal.

Return type:

dict[str, Any]

async serve(*, limit_max_requests=None)

Return type:

None

version: ClassVar[str] = 'unknown'

class fal.api.api.FalFastAPI(*args, **kwargs)

Bases: FastAPI

A subclass of FastAPI that adds some fal-specific functionality.

add_websocket_route_with_metadata(signature, endpoint)

Add websocket route and store metadata for OpenAPI injection.

openapi()

Build the OpenAPI specification for the served function. Attach needed metadata for a better integration to fal.

Return type:

dict[str, Any]

exception fal.api.api.FalMissingDependencyError(message)

Bases: FalSerializationError

exception fal.api.api.FalSerializationError(message)

Bases: FalServerlessException

message: str

class fal.api.api.FalServer(config)

Bases: Server

handle_exit(sig, frame)

set_handle_exit(handle_exit)

exception fal.api.api.FalServerlessError(message)

Bases: FalServerlessException

message: str

class fal.api.api.FalServerlessHost(url='api.alpha.fal.ai', local_file_path='', credentials=<factory>, environment_name=None)

Bases: Host

credentials: Credentials

environment_name: Optional[str] = None

files_sync(options)

Sync files to the server.

Return type:

list[File]

local_file_path: str = ''

register(func, options, *, application_name=None, application_auth_mode=None, source_code=None, metadata=None, deployment_strategy, scale=True, environment_name=None)

Return type:

Optional[RegisterApplicationResult]

run(func, options, args, kwargs, application_name=None, application_auth_mode=None)

Run the given function in the isolated environment.

Return type:

TypeVar(ReturnT, covariant=True)

spawn(func, options, args, kwargs, application_name=None, application_auth_mode=None)

Return type:

SpawnInfo

url: str = 'api.alpha.fal.ai'

class fal.api.api.Host

Bases: Generic[ArgsT, ReturnT]

The physical environment where the isolated code is executed.

classmethod parse_key(key, value)

Return type:

tuple[Any, Any]

classmethod parse_options(**config)

Split the given set of options into host and environment options.

Return type:

Options

run(func, options, args, kwargs, application_name=None, application_auth_mode=None)

Run the given function in the isolated environment.

Return type:

TypeVar(ReturnT, covariant=True)

spawn(func, options, args, kwargs, application_name=None, application_auth_mode=None)

Return type:

SpawnInfo

exception fal.api.api.InternalFalServerlessError(message)

Bases: FalServerlessException

message: str

class fal.api.api.IsolatedFunction(host, raw_func, options, executor=<factory>, reraise=True, app_name=None, app_auth=None)

Bases: Generic[ArgsT, ReturnT]

app_auth: Optional[Literal['public', 'private', 'shared']] = None

app_name: str | None = None

executor: ThreadPoolExecutor

property func: Callable[[ArgsT], ReturnT]

host: Host[ParamSpec(ArgsT), TypeVar(ReturnT, covariant=True)]

on(host=None, **config)

options: Options

raw_func: Callable[[ParamSpec(ArgsT)], TypeVar(ReturnT, covariant=True)]

reraise: bool = True

run_local(*args, **kwargs)

Return type:

TypeVar(ReturnT, covariant=True)

spawn(*args, **kwargs)

submit(*args, **kwargs)

class fal.api.api.LocalHost(_AGENT_ENVIRONMENT=<factory>)

Bases: Host

run(func, options, args, kwargs, application_name=None, application_auth_mode=None)

Run the given function in the isolated environment.

Return type:

TypeVar(ReturnT, covariant=True)

class fal.api.api.Options(host=<factory>, environment=<factory>, gateway=<factory>)

Bases: object

add_requirements(requirements)

environment: Dict[str, Any]

gateway: Dict[str, Any]

get_exposed_port()

Return type:

int | None

host: Dict[str, Any]

class fal.api.api.RouteSignature(path, is_websocket, health_check, input_modal, output_modal, realtime_mode, content_type, buffering, session_timeout, max_batch_size, emit_timings, encode_message, decode_message)

Bases: NamedTuple

buffering: int | None

Alias for field number 7

content_type: str | None

Alias for field number 6

decode_message: Optional[Callable[[bytes], Any]]

Alias for field number 12

emit_timings: bool

Alias for field number 10

encode_message: Optional[Callable[[Any], bytes]]

Alias for field number 11

health_check: HealthCheck | None

Alias for field number 2

input_modal: type | None

Alias for field number 3

is_websocket: bool

Alias for field number 1

max_batch_size: int

Alias for field number 9

output_modal: type | None

Alias for field number 4

path: str

Alias for field number 0

realtime_mode: str | None

Alias for field number 5

session_timeout: float | None

Alias for field number 8

class fal.api.api.ServeWrapper(func)

Bases: BaseServable

collect_routes()

Return type:

dict[RouteSignature, Callable[..., Any]]

class fal.api.api.ServedIsolatedFunction(host, raw_func, options, executor=<factory>, reraise=True, app_name=None, app_auth=None)

Bases: Generic[ArgsT, ReturnT], IsolatedFunction[(), None]

on(**kwds)

Helper for @overload to raise when called.

class fal.api.api.SpawnInfo(future=None, logs=<factory>, _url_ready=<factory>, _url=None, stream=None)

Bases: object

property application

future: Future | None = None

logs: Queue

property return_value

stream: Any = None

property url

exception fal.api.api.UserFunctionException

Bases: FalServerlessException

fal.api.api.cached(func)

Cache the result of the given function in-memory.

Return type:

Callable[[ParamSpec(ArgsT)], TypeVar(ReturnT, covariant=True)]

fal.api.api.find_missing_dependencies(func, env)

Return type:

Iterator[tuple[str, list[str]]]

fal.api.api.function(kind='virtualenv', *, host=None, local_python_modules=None, **config)

fal.api.api.merge_basic_config(target, incoming)

Return type:

None

fal.api.apps module

fal.api.apps.apps_runners(client, app_name, *, since=None, state=None, environment_name=None)

Return type:

List[RunnerInfo]

fal.api.apps.list_apps(client, *, filter=None, environment_name=None)

Return type:

List[AliasInfo]

fal.api.apps.rollout_app(client, app_name, *, force=False, environment_name=None)

Return type:

None

fal.api.apps.scale_app(client, app_name, *, keep_alive=None, max_multiplexing=None, max_concurrency=None, min_concurrency=None, concurrency_buffer=None, concurrency_buffer_perc=None, scaling_delay=None, request_timeout=None, startup_timeout=None, machine_types=None, regions=None, environment_name=None)

Return type:

AliasInfo

fal.api.client module

class fal.api.client.AppsNamespace(client)

Bases: object

Namespace for app management operations.

Corresponds to fal apps … CLI commands.

Accessed via client.apps.

Example

from fal.api import SyncServerlessClient

client = SyncServerlessClient() apps = client.apps.list() client.apps.scale(“my-app”, max_concurrency=10)

list(*, filter=None, environment_name=None)

List all applications. Corresponds to fal apps list.

Parameters:

• filter (str | None) – Optional app name filter string.

• environment_name (str | None) – Optional environment name.

Return type:

List[AliasInfo]

Example

apps = client.apps.list() filtered = client.apps.list(filter=”stable”)

rollout(app_name, *, force=False, environment_name=None)

Return type:

None

runners(app_name, *, since=None, state=None, environment_name=None)

List runners for a specific app. Corresponds to fal apps runners <app>.

Parameters:

• app_name (str) – Name of the application.

• since – Only return runners started after this datetime.

• state (Optional[List[str]]) – Filter by runner state (e.g., [“running”]).

• environment_name (str | None) – Optional environment name.

Return type:

List[RunnerInfo]

Example

from datetime import datetime, timedelta

runners = client.apps.runners(“my-app”) recent = client.apps.runners(

“my-app”, since=datetime.now() - timedelta(hours=1)

) running = client.apps.runners(“my-app”, state=[“running”])

scale(app_name, *, keep_alive=None, max_multiplexing=None, max_concurrency=None, min_concurrency=None, concurrency_buffer=None, concurrency_buffer_perc=None, scaling_delay=None, request_timeout=None, startup_timeout=None, machine_types=None, regions=None, environment_name=None)

Adjust scaling settings for an application. Corresponds to fal apps scale.

Any omitted option keeps the current value.

Parameters:

• app_name (str) – Name of the application.

• keep_alive (int | None) – Keep-alive time in seconds.

• max_multiplexing (int | None) – Maximum request multiplexing.

• max_concurrency (int | None) – Maximum concurrent runners.

• min_concurrency (int | None) – Minimum concurrent runners.

• request_timeout (int | None) – Request timeout in seconds.

• startup_timeout (int | None) – Startup timeout in seconds.

• machine_types (Optional[List[str]]) – List of allowed machine types (e.g., [“GPU-H100”]).

• regions (Optional[List[str]]) – List of allowed regions (e.g., [“us-east”]).

Return type:

AliasInfo

Example

client.apps.scale(

“my-app”, keep_alive=300, max_concurrency=10, min_concurrency=1, machine_types=[“GPU-H100”, “GPU-H200”],

)

class fal.api.client.KeysNamespace(client)

Bases: object

Namespace for API key management. Corresponds to fal keys … CLI commands.

Accessed via client.keys.

Example

from fal.api import SyncServerlessClient

client = SyncServerlessClient() keys = client.keys.list() key_id, key_secret = client.keys.create(scope=”admin”)

create(*, scope, description=None)

Create a new API key.

Parameters:

• scope (KeyScope) – Key scope (e.g., “admin”).

• description (str | None) – Optional description for the key.

Return type:

tuple[str, str]

Returns:

Tuple of (key_id, key_secret).

list()

List all API keys.

Return type:

List[UserKeyInfo]

revoke(key_id)

Revoke an API key.

Parameters:

key_id (str) – The ID of the key to revoke.

Return type:

None

class fal.api.client.RunnersNamespace(client)

Bases: object

Namespace for runner management operations.

Corresponds to fal runners … CLI commands.

Accessed via client.runners.

Example

from fal.api import SyncServerlessClient

client = SyncServerlessClient() runners = client.runners.list() client.runners.stop(“runner-id”)

kill(runner_id)

Forcefully kill a runner.

Parameters:

runner_id (str) – The ID of the runner to kill.

Return type:

None

list(*, since=None)

List all runners. Corresponds to fal runners list.

Parameters:

since – Only return runners started after this datetime.

Return type:

List[RunnerInfo]

Example

from datetime import datetime, timedelta

all_runners = client.runners.list() recent = client.runners.list(since=datetime.now() - timedelta(minutes=10))

stop(runner_id, replace_first=False)

Gracefully stop a runner.

Parameters:

• runner_id (str) – The ID of the runner to stop.

• replace_first (bool) – Whether to replace the runner before stopping it.

Return type:

None

class fal.api.client.SecretsNamespace(client)

Bases: object

Namespace for secrets management. Corresponds to fal secrets … CLI commands.

Accessed via client.secrets.

Example

from fal.api import SyncServerlessClient

client = SyncServerlessClient() client.secrets.set(“API_KEY”, “my-secret-value”) secrets = client.secrets.list()

list(environment_name=None)

List all secrets (names only, not values).

Return type:

List[ServerlessSecret]

set(name, value, environment_name=None)

Set a secret value.

Parameters:

• name (str) – Name of the secret.

• value (str) – Value to store.

Return type:

None

unset(name, environment_name=None)

Delete a secret.

Parameters:

• name (str) – Name of the secret to delete.

• environment_name (str | None) – Optional environment name.

Return type:

None

class fal.api.client.SyncServerlessClient(host=None, api_key=None, profile=None, team=None)

Bases: object

Synchronous Python client for fal Serverless.

Manage apps, runners, and deployments programmatically. The namespaces and methods mirror the CLI so you can automate the same workflows from Python.

Parameters:

• host (Optional[str]) – Optional. Override API host.

• api_key (Optional[str]) – Optional. If omitted, read from env/profile.

• profile (Optional[str]) – Optional. Named profile to use.

• team (Optional[str]) – Optional. Team context for runner operations.

Example

from fal.api import SyncServerlessClient

client = SyncServerlessClient()

# List apps apps = client.apps.list()

# Scale an app client.apps.scale(“my-app”, max_concurrency=10)

# Deploy client.deploy(“path/to/myfile.py::MyApp”)

Namespaces:

• client.apps.* - corresponds to fal apps …

• client.runners.* - corresponds to fal runners …

• client.keys.* - corresponds to fal keys …

• client.secrets.* - corresponds to fal secrets …

• client.deploy() - corresponds to fal deploy …

api_key: Optional[str] = None

deploy(*args, **kwargs)

Deploy an application. Corresponds to fal deploy.

If app_ref is omitted, discovery behavior matches the CLI (e.g., uses pyproject.toml).

Parameters:

• app_ref – Path to file, file::Class, or existing app name.

• app_name – Override the application name.

• auth – Authentication mode (“private” | “public”).

• strategy – Deployment strategy (“recreate” | “rolling”).

• reset_scale – If False, use previous scaling settings.

Example

# Auto-discover from pyproject.toml client.deploy()

# Deploy from a file path client.deploy(“path/to/myfile.py”)

# Deploy a specific class client.deploy(“path/to/myfile.py::MyApp”)

# With options client.deploy(

app_ref=”path/to/myfile.py::MyApp”, app_name=”myapp”, auth=”public”, strategy=”rolling”,

)

host: Optional[str] = None

profile: Optional[str] = None

team: Optional[str] = None

fal.api.deploy module

class fal.api.deploy.DeploymentResult(revision, app_name, urls, log_url, auth_mode)

Bases: object

app_name: str

auth_mode: str

log_url: str

revision: str

urls: dict[str, dict[str, str]]

class fal.api.deploy.User(user_id, username)

Bases: tuple

user_id

Alias for field number 0

username

Alias for field number 1

fal.api.deploy.deploy(client, app_ref=None, *, app_name=None, auth=None, strategy='rolling', reset_scale=False, force_env_build=False, environment_name=None)

Return type:

DeploymentResult

fal.api.environments module

fal.api.environments.create_environment(client, name, description=None)

Return type:

EnvironmentInfo

fal.api.environments.delete_environment(client, name)

Return type:

None

fal.api.environments.list_environments(client)

Return type:

List[EnvironmentInfo]

fal.api.keys module

fal.api.keys.create_key(client, *, scope, description=None)

Return type:

tuple[str, str]

fal.api.keys.list_keys(client)

Return type:

List[UserKeyInfo]

fal.api.keys.revoke_key(client, key_id)

Return type:

None

fal.api.runners module

fal.api.runners.kill_runner(client, runner_id)

Return type:

None

fal.api.runners.list_runners(client, *, since=None)

Return type:

List[RunnerInfo]

fal.api.runners.stop_runner(client, runner_id, replace_first=False)

Return type:

None

fal.api.secrets module

fal.api.secrets.list_secrets(client, environment_name=None)

Return type:

List[ServerlessSecret]

fal.api.secrets.set_secret(client, name, value, environment_name=None)

Return type:

None

fal.api.secrets.unset_secret(client, name, environment_name=None)

Return type:

None

Module contents

class fal.api.AppsNamespace(client)

Bases: object

Namespace for app management operations.

Corresponds to fal apps … CLI commands.

Accessed via client.apps.

Example

from fal.api import SyncServerlessClient

client = SyncServerlessClient() apps = client.apps.list() client.apps.scale(“my-app”, max_concurrency=10)

list(*, filter=None, environment_name=None)

List all applications. Corresponds to fal apps list.

Parameters:

• filter (str | None) – Optional app name filter string.

• environment_name (str | None) – Optional environment name.

Return type:

List[AliasInfo]

Example

apps = client.apps.list() filtered = client.apps.list(filter=”stable”)

rollout(app_name, *, force=False, environment_name=None)

Return type:

None

runners(app_name, *, since=None, state=None, environment_name=None)

List runners for a specific app. Corresponds to fal apps runners <app>.

Parameters:

• app_name (str) – Name of the application.

• since – Only return runners started after this datetime.

• state (Optional[List[str]]) – Filter by runner state (e.g., [“running”]).

• environment_name (str | None) – Optional environment name.

Return type:

List[RunnerInfo]

Example

from datetime import datetime, timedelta

runners = client.apps.runners(“my-app”) recent = client.apps.runners(

“my-app”, since=datetime.now() - timedelta(hours=1)

) running = client.apps.runners(“my-app”, state=[“running”])

scale(app_name, *, keep_alive=None, max_multiplexing=None, max_concurrency=None, min_concurrency=None, concurrency_buffer=None, concurrency_buffer_perc=None, scaling_delay=None, request_timeout=None, startup_timeout=None, machine_types=None, regions=None, environment_name=None)

Adjust scaling settings for an application. Corresponds to fal apps scale.

Any omitted option keeps the current value.

Parameters:

• app_name (str) – Name of the application.

• keep_alive (int | None) – Keep-alive time in seconds.

• max_multiplexing (int | None) – Maximum request multiplexing.

• max_concurrency (int | None) – Maximum concurrent runners.

• min_concurrency (int | None) – Minimum concurrent runners.

• request_timeout (int | None) – Request timeout in seconds.

• startup_timeout (int | None) – Startup timeout in seconds.

• machine_types (Optional[List[str]]) – List of allowed machine types (e.g., [“GPU-H100”]).

• regions (Optional[List[str]]) – List of allowed regions (e.g., [“us-east”]).

Return type:

AliasInfo

Example

client.apps.scale(

“my-app”, keep_alive=300, max_concurrency=10, min_concurrency=1, machine_types=[“GPU-H100”, “GPU-H200”],

)

class fal.api.KeysNamespace(client)

Bases: object

Namespace for API key management. Corresponds to fal keys … CLI commands.

Accessed via client.keys.

Example

from fal.api import SyncServerlessClient

client = SyncServerlessClient() keys = client.keys.list() key_id, key_secret = client.keys.create(scope=”admin”)

create(*, scope, description=None)

Create a new API key.

Parameters:

• scope (KeyScope) – Key scope (e.g., “admin”).

• description (str | None) – Optional description for the key.

Return type:

tuple[str, str]

Returns:

Tuple of (key_id, key_secret).

list()

List all API keys.

Return type:

List[UserKeyInfo]

revoke(key_id)

Revoke an API key.

Parameters:

key_id (str) – The ID of the key to revoke.

Return type:

None

class fal.api.RunnersNamespace(client)

Bases: object

Namespace for runner management operations.

Corresponds to fal runners … CLI commands.

Accessed via client.runners.

Example

from fal.api import SyncServerlessClient

client = SyncServerlessClient() runners = client.runners.list() client.runners.stop(“runner-id”)

kill(runner_id)

Forcefully kill a runner.

Parameters:

runner_id (str) – The ID of the runner to kill.

Return type:

None

list(*, since=None)

List all runners. Corresponds to fal runners list.

Parameters:

since – Only return runners started after this datetime.

Return type:

List[RunnerInfo]

Example

from datetime import datetime, timedelta

all_runners = client.runners.list() recent = client.runners.list(since=datetime.now() - timedelta(minutes=10))

stop(runner_id, replace_first=False)

Gracefully stop a runner.

Parameters:

• runner_id (str) – The ID of the runner to stop.

• replace_first (bool) – Whether to replace the runner before stopping it.

Return type:

None

class fal.api.SecretsNamespace(client)

Bases: object

Namespace for secrets management. Corresponds to fal secrets … CLI commands.

Accessed via client.secrets.

Example

from fal.api import SyncServerlessClient

client = SyncServerlessClient() client.secrets.set(“API_KEY”, “my-secret-value”) secrets = client.secrets.list()

list(environment_name=None)

List all secrets (names only, not values).

Return type:

List[ServerlessSecret]

set(name, value, environment_name=None)

Set a secret value.

Parameters:

• name (str) – Name of the secret.

• value (str) – Value to store.

Return type:

None

unset(name, environment_name=None)

Delete a secret.

Parameters:

• name (str) – Name of the secret to delete.

• environment_name (str | None) – Optional environment name.

Return type:

None

class fal.api.SyncServerlessClient(host=None, api_key=None, profile=None, team=None)

Bases: object

Synchronous Python client for fal Serverless.

Manage apps, runners, and deployments programmatically. The namespaces and methods mirror the CLI so you can automate the same workflows from Python.

Parameters:

• host (Optional[str]) – Optional. Override API host.

• api_key (Optional[str]) – Optional. If omitted, read from env/profile.

• profile (Optional[str]) – Optional. Named profile to use.

• team (Optional[str]) – Optional. Team context for runner operations.

Example

from fal.api import SyncServerlessClient

client = SyncServerlessClient()

# List apps apps = client.apps.list()

# Scale an app client.apps.scale(“my-app”, max_concurrency=10)

# Deploy client.deploy(“path/to/myfile.py::MyApp”)

Namespaces:

• client.apps.* - corresponds to fal apps …

• client.runners.* - corresponds to fal runners …

• client.keys.* - corresponds to fal keys …

• client.secrets.* - corresponds to fal secrets …

• client.deploy() - corresponds to fal deploy …

api_key: Optional[str] = None

deploy(*args, **kwargs)

Deploy an application. Corresponds to fal deploy.

If app_ref is omitted, discovery behavior matches the CLI (e.g., uses pyproject.toml).

Parameters:

• app_ref – Path to file, file::Class, or existing app name.

• app_name – Override the application name.

• auth – Authentication mode (“private” | “public”).

• strategy – Deployment strategy (“recreate” | “rolling”).

• reset_scale – If False, use previous scaling settings.

Example

# Auto-discover from pyproject.toml client.deploy()

# Deploy from a file path client.deploy(“path/to/myfile.py”)

# Deploy a specific class client.deploy(“path/to/myfile.py::MyApp”)

# With options client.deploy(

app_ref=”path/to/myfile.py::MyApp”, app_name=”myapp”, auth=”public”, strategy=”rolling”,

)

host: Optional[str] = None

profile: Optional[str] = None

team: Optional[str] = None