Programmatic Deployment API Reference

Programmatic Deployer API

Class: `AppDeployer`

AppDeployer(
    name: Optional[str] = None,
    port: Optional[int] = None,
    description: Optional[str] = None,
    app_type: Optional[str] = None,
    image: Optional[str] = None,
    tags: Optional[list] = None,
    secrets: Optional[list] = None,
    compute_pools: Optional[list] = None,
    environment: Optional[dict] = None,
    commands: Optional[list] = None,
    resources: Optional[ResourceConfigDict] = None,
    auth: Optional[AuthConfigDict] = None,
    replicas: Optional[ReplicaConfigDict] = None,
    code_package: Optional[tuple] = None,
    force_upgrade: Optional[bool] = None,
    persistence: Optional[str] = None,
    project: Optional[str] = None,
    branch: Optional[str] = None,
    models: Optional[list] = None,
    data: Optional[list] = None,
    generate_static_url: Optional[bool] = None,
    **kwargs
)

Module: metaflow.apps

Programmatic API For deploying Outerbounds Apps.

Parameters

name (str, optional): The name of the app to deploy.
port (int, optional): Port where the app is hosted. When deployed this will be port on which we will deploy the app.
description (str, optional): The description of the app to deploy.
app_type (str, optional): The User defined type of app to deploy. Its only used for bookkeeping purposes.
image (str, optional): The Docker image to deploy with the App.
tags (list, optional): The tags of the app to deploy.
secrets (list, optional): Outerbounds integrations to attach to the app. You can use the value you set in the @secrets decorator in your code.
compute_pools (list, optional): A list of compute pools to deploy the app to.
environment (dict, optional): Environment variables to deploy with the App.
commands (list, optional): A list of commands to run the app with.
resources (ResourceConfigDict, optional): Resource configuration for the app.
- cpu (str) CPU requests
- memory (str) Memory requests
- gpu (str) GPU requests
- disk (str) Storage disk size.
- shared_memory (str) Shared memory
auth (AuthConfigDict, optional): Auth related configurations.
- type (str) The type of authentication to use for the app.
- public (bool) Whether the app is public or not.
replicas (ReplicaConfigDict, optional): The number of replicas to deploy the app with.
- fixed (int) The fixed number of replicas to deploy the app with. If min and max are set, this will raise an error.
- min (int) The minimum number of replicas to deploy the app with.
- max (int) The maximum number of replicas to deploy the app with.
- scaling_policy (ScalingPolicyConfigDict) Scaling policy defines the the metric based on which the replicas will horizontally scale. If min and max replicas are set and are not the same, then a scaling policy will be applied. Default scaling policies can be 60 rpm (ie 1 rps).
  - rpm (int) Scale up replicas when the requests per minute crosses this threshold. If nothing is provided and the replicas.max and replicas.min is set then the default rpm would be 60.
code_package (tuple, optional): Pre-packaged code from package_code(). A PackagedCode namedtuple containing url and key.
force_upgrade (bool, optional): Force upgrade the app even if it is currently being upgraded.
persistence (str, optional): The persistence mode to deploy the app with. [Experimental] May change in the future.
project (str, optional): The project name to deploy the app to. [Experimental] May change in the future.
branch (str, optional): The branch name to deploy the app to. [Experimental] May change in the future.
models (list, optional): [Experimental] May change in the future.
data (list, optional): [Experimental] May change in the future.
generate_static_url (bool, optional): Generate a static URL for the app based on its name.

Examples

Basic deployment with bake_image and package_code:

from metaflow.apps import bake_image, package_code, AppDeployer

# Step 1: Bake dependencies into an image
baked = bake_image(pypi={"flask": ">=2.0", "requests": ""})

# Step 2: Package your application code
pkg = package_code(src_paths=["./src"])

# Step 3: Create deployer and deploy
deployer = AppDeployer(
    name="my-flask-app",
    port=8000,
    image=baked.image,
    code_package=pkg,
    commands=["python server.py"],
    replicas={"min": 1, "max": 3},
    resources={"cpu": "1", "memory": "2048Mi"},
)
deployed = deployer.deploy()
print(deployed.public_url)

Deployment with API authentication:

deployer = AppDeployer(
    name="my-api",
    port=8000,
    image=baked.image,
    code_package=pkg,
    commands=["python api.py"],
    auth={"type": "API"},
)
deployed = deployer.deploy()

Deployment with environment variables and secrets:

deployer = AppDeployer(
    name="my-app",
    port=8000,
    image=baked.image,
    code_package=pkg,
    commands=["python app.py"],
    environment={"DEBUG": "false", "LOG_LEVEL": "info"},
    secrets=["my-api-keys"],
)
deployed = deployer.deploy()

Interacting with a deployed app:

# Get app info
info = deployed.info()

# Get logs from all workers
logs = deployed.logs()

# Scale to zero workers
deployed.scale_to_zero()

# Delete the app
deployed.delete()

Method: `AppDeployer.deploy`

AppDeployer.deploy(
    readiness_condition: str = 'at_least_one_running',
    max_wait_time = 600,
    readiness_wait_time = 10,
    logger_fn: Callable = functools.partial(<built-in function print>, fi...,
    **kwargs
)

Deploy the app to the Outerbounds Platform.

This method packages and deploys the configured app, waiting for it to reach the specified readiness condition before returning.

Parameters

readiness_condition (str, optional): The condition that must be met for the deployment to be considered ready. Default is ATLEAST_ONE_RUNNING.

Deployment ready conditions define what is considered a successful completion of the current deployment instance. This allows users or platform designers to configure the criteria for deployment readiness.

Why do we need deployment readiness conditions?

- Deployments might be taking place from a CI/CD-esque environment.
  In these setups, the downstream build triggers might be depending on
  a specific criteria for deployment completion. Having readiness conditions
  allows the CI/CD systems to get a signal of when the deployment is ready.
- Users might be calling the deployment API under different conditions:
    - Some users might want a cluster of workers ready before serving
      traffic while others might want just one worker ready to start
      serving traffic.

Available readiness conditions:

ATLEAST_ONE_RUNNING ("at_least_one_running") At least min(min_replicas, 1) workers of the current deployment instance's version have started running. Usecase: Some endpoints may be deployed ephemerally and are considered ready when at least one instance is running; additional instances are for load management.

ALL_RUNNING ("all_running") At least min_replicas number of workers are running for the deployment to be considered ready. Usecase: Operators may require that all replicas are available before traffic is routed. Needed when inference endpoints may be under some SLA or require a larger load.

FULLY_FINISHED ("fully_finished") At least min_replicas number of workers are running for the deployment and there are no pending or crashlooping workers from previous versions lying around. Usecase: Ensuring endpoint is fully available and no other versions are running or endpoint has been fully scaled down.

ASYNC ("async") The deployment will be assumed ready as soon as the server acknowledges it has registered the app in the backend. Usecase: Operators may only care that the URL is minted for the deployment or the operator wants the deployment to eventually scale down to 0.

max_wait_time (int, optional): Maximum time in seconds to wait for the deployment to reach readiness. Default is 600 (10 minutes).
readiness_wait_time (int, optional): Time in seconds to wait between readiness checks. Default is 10.
logger_fn (Callable, optional): Function to use for logging progress messages. Default prints to stderr.

Returns

*** (DeployedApp*): An object representing the deployed app with methods to interact with it (logs, info, scale_to_zero, delete, etc.) and properties like public_url.

Raises

*** (CodePackagingException*): If code_package is not provided or is not a valid PackagedCode instance.
*** (AppConfigError*): If the app configuration is invalid.
*** (AppCreationFailedException*): If the app deployment submission fails due to an API error. Contains status_code and error_text attributes for debugging.
*** (AppCrashLoopException*): If a worker enters CrashLoopBackOff or Failed state during deployment. Contains worker_id and logs attributes for debugging.
*** (AppReadinessException*): If the app fails to meet readiness conditions within max_wait_time.
*** (AppUpgradeInProgressException*): If an upgrade is already in progress when deployment starts. Use force_upgrade=True to override. Contains upgrader attribute.
*** (AppConcurrentUpgradeException*): If another deployment was triggered while this deployment was in progress, invalidating the current deployment. Contains expected_version and actual_version.
*** (OuterboundsBackendUnhealthyException*): If the Outerbounds backend is unreachable (network issues, DNS failures) or returns server errors (HTTP 5xx). This indicates a platform-side issue, not a problem with your configuration. Retry the deployment or contact Outerbounds support.
*** (AppDeletedDuringDeploymentException*): If the app was deleted by another process or user while this deployment was in progress. This can occur when concurrent operations conflict.

Examples

Basic deployment:

from metaflow.apps import bake_image, package_code, AppDeployer
baked = bake_image(pypi={"flask": ">=2.0"})
pkg = package_code(src_paths=["./src"])
deployer = AppDeployer(
    name="my-app",
    port=8000,
    image=baked.image,
    code_package=pkg,
    commands=["python server.py"],
)
deployed = deployer.deploy()
print(deployed.public_url)

Wait for all replicas to be ready:

deployed = deployer.deploy(
    readiness_condition="all_running"
)

Async deployment (don't wait for workers):

deployed = deployer.deploy(
    readiness_condition="async"
)

Handling deployment errors:

from metaflow.apps import AppDeployer
from metaflow.apps.exceptions import (
    AppReadinessException,
)

try:
    deployed = deployer.deploy()
except AppReadinessException as e:
    print(f"App {e.app_id} failed to become ready in time but we can move forward")
    deployed_app:DeployedApp = e.deployed_app
    # use DeployedApp to do what ever you need

Method: `AppDeployer.list_deployments`

AppDeployer.list_deployments(
    name: str = None,
    project: str = None,
    branch: str = None,
    tags: List[Dict[str, str]] = None
)

List deployed apps, optionally filtered by name, project, branch, or tags.

Parameters

name (str, optional): Filter by app name.
project (str, optional): Filter by project name.
branch (str, optional): Filter by branch name.
tags (List[Dict[str, str]], optional): Filter by tags. Each tag is a dict with a single key-value pair, e.g., [{"env": "prod"}] or [{"team": "ml"}, {"version": "v2"}]. Apps must have all specified tags to match.

Returns

*** (List[DeployedApp]*): List of deployed apps matching the filters.

Examples

List all apps:

apps = AppDeployer.list_deployments()

Filter by name:

apps = AppDeployer.list_deployments(name="my-app")

Filter by project and branch:

apps = AppDeployer.list_deployments(project="ml-pipeline", branch="main")

Filter by a single tag:

apps = AppDeployer.list_deployments(tags=[{"env": "prod"}])

Filter by multiple tags (AND logic - must match all):

apps = AppDeployer.list_deployments(tags=[{"env": "prod"}, {"team": "ml"}])

Combine filters:

apps = AppDeployer.list_deployments(
    project="recommendations",
    tags=[{"env": "staging"}]
)

Class: `DeployedApp`

Module: metaflow.apps

A deployed app on the Outerbounds Platform.

Obtain instances via AppDeployer.deploy() or AppDeployer.list_deployments().

Examples

After deployment:

deployed = deployer.deploy()
print(deployed.public_url)

After listing:

apps = AppDeployer.list_deployments(tags=[{"env": "staging"}])
for app in apps:
    print(f"{app.name}: {app.public_url}")

Inspect and manage:

# Get logs
for worker_id, lines in deployed.logs().items():
    print(f"Worker {worker_id}: {len(lines)} log lines")

# Scale down
deployed.scale_to_zero()

# Clean up
deployed.delete()

Make authenticated requests (API auth):

import requests
response = requests.get(deployed.public_url, headers=deployed.auth())

Property: `DeployedApp.id`

Unique identifier for the deployed app.

Returns

str: The unique app identifier assigned by the platform.

Property: `DeployedApp.auth_type`

Authentication type configured for this app. Can be either Browser , API, BrowserAndApi

Returns

str: The authentication type

Property: `DeployedApp.public_url`

Public URL to access the deployed app.

Returns

str: The publicly accessible URL for this app.

Property: `DeployedApp.internal_url`

Internal in-cluster URL to access the deployed app.

This URL bypasses external network routing and can be used from within Metaflow tasks running on Kubernetes. Authentication headers are not required when accessing the app via this URL from within the cluster.

Returns

str: The in-cluster URL for this app.

Property: `DeployedApp.name`

Logical name given to the app.

Returns

str: The human-readable name of the app.

Property: `DeployedApp.deployed_version`

Current deployment version of the app.

Returns

str: The version identifier for the current deployment.

Property: `DeployedApp.tags`

Tags associated with this app.

Returns

List[str]: List of tags assigned to this app.

Method: `DeployedApp.info`

DeployedApp.info()

Get detailed information about the deployed app.

Returns

*** (dict*): Dictionary containing full app details including spec, status, metadata, and configuration.

Examples

info = deployed.info()
print(f"Status: {info.get('status')}")
print(f"Spec: {info.get('spec')}")

Method: `DeployedApp.logs`

DeployedApp.logs(previous: bool = False)

Get logs from all worker replicas.

Parameters

previous (bool, optional): If True, returns logs from the previous execution of workers. Useful for debugging crashlooping workers. Default is False.

Returns

*** (Dict[str, List[LogLine]]*): Dictionary mapping worker IDs to their log lines.

Examples

# Get current logs
logs = deployed.logs()
for worker_id, lines in logs.items():
    print(f"Worker {worker_id}:")
    for line in lines:
        print(f"  {line}")

# Get logs from crashed workers
previous_logs = deployed.logs(previous=True)

Method: `DeployedApp.replicas`

DeployedApp.replicas()

List all active worker replicas for this app.

Returns

*** (List[dict]*): List of dictionaries containing worker information including workerId, status, and other metadata.

Examples

workers = deployed.replicas()
for worker in workers:
    print(f"Worker {worker['workerId']}: {worker.get('status')}")

Method: `DeployedApp.scale_to_zero`

DeployedApp.scale_to_zero()

Scale the app down to zero replicas.

This stops all running workers while preserving the app configuration. The app can be scaled back up by sending traffic to the public URL (if autoscaling is configured) or by redeploying.

Examples

# Scale down to save resources
deployed.scale_to_zero()

Method: `DeployedApp.delete`

DeployedApp.delete()

Delete the deployed app.

This permanently removes the app from the platform, including all workers, configuration, and the public URL. This action cannot be undone.

Examples

# Clean up the app
deployed.delete()

Method: `DeployedApp.auth`

DeployedApp.auth()

Get authentication headers for making requests to this app.

Only available for apps configured with API authentication type. Use these headers when making HTTP requests to the app's public URL.

Returns

*** (dict*): Dictionary of HTTP headers to include in requests.

Raises

*** (ValueError*): If the app is not configured with API authentication.