SDK

Install

During early development, install the toolkit directly from GitHub into a clean uv environment:

uv venv .venv
source .venv/bin/activate
uv pip install "rebase-toolkit @ git+ssh://git@github.com/rebase-energy/rebase-toolkit.git"

Once the package is published to PyPI, install from PyPI:

pip install rebase-toolkit

Then authenticate this computer and select or create a workspace:

rebase setup

rebase setup signs you in with Google or GitHub and stores a local auth session and workspace profile on your computer. The hosted Rebase API URL is built into the SDK, so normal user code does not need an API URL, token, or API key argument.

Configure

Most code should just import rebase and use the default client:

import rebase as rb

project = rb.project("energy-forecasting")

For multiple workspace credentials, use CLI profiles:

rebase setup --profile prod
rebase workspace switch prod

The lower-level Client class exists for platform internals and tests, but it is not the normal user-facing API.

Project-First Registration

Use rb.project(...) when registering new targets. A project can contain reusable functions and pipeline-style workflows.

import rebase as rb

project = rb.project("energy-forecasting")

@project.function(name="add-numbers")
def add(left: float = 0, right: float = 0) -> dict:
    return {"sum": left + right}

@project.step(name="load-weather", retries=2)
def load_weather(site_id: str) -> dict:
    return {"site_id": site_id, "temperature_c": 11.2}

@project.workflow(name="daily-forecast")
def forecast(site_id: str) -> dict:
    weather = load_weather(site_id)
    return {"weather": weather}

project.deploy()

Functions use run_type="quick" by default, which runs each call synchronously on an isolated Cloud Run service that stays warm after the first call. Use run_type="long" when a function should run as a Cloud Run Job, and run_type="quick_shared" for the shared runner path. See Function Run Types for the isolation model and latency shape.

@project.function(name="quick-task", min_instances=1, concurrency=1)
def quick_task(site_id: str) -> dict:
    return {"site_id": site_id}

@project.function(name="long-task", run_type="long")
def long_task(site_id: str) -> dict:
    return {"site_id": site_id}

@project.function(name="shared-task", run_type="quick_shared")
def shared_task(site_id: str) -> dict:
    return {"site_id": site_id}

For quick functions, Rebase creates a private Cloud Run service for the function and one tagged revision per immutable function version. concurrency=1 prevents concurrent user calls inside the same Cloud Run instance. min_instances=1 keeps the function warm, but creates steady Cloud Run cost.

The old backend= argument is removed and raises an error with a migration hint: backend="interactive" becomes run_type="quick", backend="batch" becomes run_type="long".

Workflows also default to run_type="quick", which resolves to a warm Cloud Run service worker backed by a Cloud Run-hosted Prefect API. Set run_type="long" when each workflow run should execute as a Cloud Run Job. See Workflow Run Types.

Add public PyPI dependencies with dependencies=[...]. Rebase installs them with uv pip install in a fingerprinted cache for the function version:

@project.function(name="mean-load", dependencies=["numpy==2.3.0"])
def mean_load(values: list[float]) -> dict:
    import numpy as np

    return {"mean": float(np.mean(values))}

Use rb.Image when you want the image spec to be explicit:

image = rb.Image.python("3.13").uv_pip_install("pandas==2.3.0")

@project.function(name="daily-summary", image=image)
def daily_summary(rows: list[dict]) -> dict:
    import pandas as pd

    return {"rows": int(len(pd.DataFrame(rows)))}

Pin exact dependency versions for reproducible function versions. V1 dependency support accepts public PyPI package specs only. Provide either dependencies=[...] or image=..., not both.

Local, Ephemeral, and Deployed Runs

Calling a Rebase-decorated function or workflow directly runs normal local Python:

print(add(left=3, right=4))
print(forecast(site_id="site-001"))

Run the same source in the cloud without creating a reusable target:

rebase run workflow.py --param site_id=site-001

Use project.deploy() only when you want persistent, versioned functions or workflows, or when you want to activate a scheduled workflow. Use rebase.deploy(model) or model.deploy() for persistent model targets.

Deploys target the dev environment by default. Pass environment= when deploying through the SDK:

project.deploy(environment="dev")
rebase.deploy(model, environment="staging")
model.deploy(environment="dev")

Workspace policies decide whether an environment allows direct deploys. New workspaces default to direct deploys for dev and GitOps-protected deploys for staging and prod. SDK .deploy(environment="prod") is rejected by the API when prod is protected; use rebase deploy deploy.py --env prod so Rebase can create a GitOps deployment request from committed Git source.

Deploy behavior:

HTTP Endpoints

Use rb.endpoint(...) when a deployed function, workflow, or model should also be callable over HTTP:

@project.function(
    name="add-numbers",
    endpoint=rb.endpoint(method="POST", path="/add"),
)
def add_numbers(a: float = 0, b: float = 0) -> dict:
    return {"sum": a + b}

Endpoints are deployed with the target:

rebase deploy workflow.py
rebase endpoint list --project energy-forecasting

For standalone targets, @rb.endpoint works with @rb.function and @rb.workflow. If you omit project, the target belongs to the implicit default project:

@rb.function(project="energy-forecasting")
@rb.endpoint(method="POST", path="/forecast")
def forecast(zone: str = "SE3") -> dict:
    return {"zone": zone}

Model instances accept the same endpoint config:

model = PricePredictor(
    project="energy-forecasting",
    endpoint=rb.endpoint(method="POST", path="/predict"),
)

Functions and models default to synchronous endpoint responses. Workflows default to asynchronous endpoint responses. Endpoint invocation defaults to auth="api_key" for agent and service use cases; use auth="workspace" when signed-in workspace users should invoke the endpoint through their normal rebase setup credentials.

See Endpoints for route shape, auth modes, and CLI commands.

CLI Deploy

Use the CLI after rebase setup when the file contains a top-level rb.project(...):

rebase deploy workflow.py

The CLI imports the file and deploys discovered Rebase objects. If the file contains one or more top-level projects, it deploys those projects. If there is no project, it deploys top-level standalone @rb.workflow(...) or @rb.function(...) handles.

You can deploy one target by top-level variable name or Rebase target name:

rebase deploy workflow.py --name energy-forecasting

Deploy to a specific environment:

rebase deploy workflow.py --env dev
rebase deploy workflow.py --env prod

rebase deploy reads the workspace environment policy before importing the file. Direct environments deploy immediately. GitOps-protected environments require a clean committed file in the connected GitHub repository and create a GitOps deployment request instead of mutating live state directly.

Preview the path without creating a deployment request:

rebase deploy workflow.py --env prod --plan

See GitOps Deployments for environment policies and protected deploy behavior.

Source Backing

New workspaces default to Rebase-hosted source. No GitHub repo is required:

project = rb.project("energy-forecasting")

For a workspace repo layout, configure the workspace once, then set a project folder:

rb.update_workspace(
    source_mode="workspace_repo",
    repo_owner="rebase-energy",
    repo_name="platform-workflows",
)

project = rb.project(
    "energy-forecasting",
    source_mode="workspace_repo",
    repo_path="projects/energy-forecasting",
)

When deploy runs inside a local Git repo, the SDK records commit, branch, source path, and dirty state when available. Rebase still stores the source snapshot on the immutable version.

Step Workflows

Use @project.step(...) for functions that should become nodes in a workflow graph.

@project.step()
def load_weather(site_id: str) -> dict:
    return {"site_id": site_id, "temperature_c": 11.2}

@project.step()
def load_prices(zone: str) -> dict:
    return {"zone": zone, "day_ahead_eur_mwh": 42.0}

@project.step()
def build_forecast(weather: dict, prices: dict, horizon_hours: int = 24) -> dict:
    return {
        "mw": 42,
        "weather": weather,
        "prices": prices,
        "horizon_hours": horizon_hours,
    }

@project.workflow(
    name="site-forecast",
    schedule=rb.Cron("0 6 * * *", timezone="Europe/Stockholm"),
    run_type="long",
)
def forecast(site_id: str = "site-001", zone: str = "SE3", horizon_hours: int = 24) -> dict:
    weather = load_weather(site_id)
    prices = load_prices(zone)
    forecast_result = build_forecast(weather, prices, horizon_hours=horizon_hours)
    return {"forecast": forecast_result}

project.deploy() registers the steps as functions, stores a workflow graph that pins each node to the deployed function version, and activates the cron schedule. Runtime branching should live inside a step function; the workflow body should be a static composition of step calls.

@project.step(...) registers step bodies as internal workflow graph nodes. Step decorators do not take a run_type argument: steps execute in-flow within the workflow's run, so the workflow's run type covers every step.

Cron schedules use standard five-field cron syntax:

@project.workflow(schedule=rb.Cron("0 8 * * 1", timezone="UTC"))
def weekly_forecast(site_id: str = "site-001") -> dict:
    return {"site_id": site_id, "status": "ok"}

Cron schedules only define timing. Scheduled workflows must define Python defaults for every workflow parameter; otherwise rebase deploy workflow.py fails before creating the schedule.

Call a Function

run = add.spawn(left=3, right=4)
print(run.status)

result = add.remote(left=3, right=4)
print(result)

spawn(...) returns immediately with a Run. remote(...) waits for the terminal result.

Runs expose the selected run_type plus the concrete internal execution_backend and, when the infrastructure provides one, a backend_run_id (such as a Prefect flow run ID) for debugging. Rebase remains the source of truth for the function version, run status, and normalized result.

Call by Name

Use rb.get_function(...) and rb.get_workflow(...) when code in one project needs to call a function or workflow from another project.

normalize_weather = rb.get_function("shared-energy-utils/normalize-weather")

result = normalize_weather.remote(
    temperature_c=12.5,
    wind_mps=4.2,
)

Workflow handles use the same pattern:

forecast = rb.get_workflow("energy-forecasting/daily-forecast")
run = forecast.spawn(site_id="site-001", horizon_hours=48)

Runs, Steps, Tasks, and Artifacts

run = forecast.spawn(site_id="site-001", horizon_hours=48)

for event in run.events():
    print(event["status"], event["message"])

for step in run.steps():
    print(step["name"], step["status"], step["attempt"])

run.events() returns persisted run events. run.steps() returns the step records for a workflow run. It is most useful for step workflows, where each step maps to a Prefect task run.

Below steps there is a third layer: tasks. A step that fans work out with Function.map records one task per item, each with its own status, parameters, and result or error — so a step that reports one failure can still tell you which of its thousand calls caused it:

client = rb.Client()

for task in client.list_run_tasks(run.id):
    print(task["status"], task.get("error"))

# Or just one step's fan-out:
tasks = client.list_run_tasks(run.id, step_run_id=step["id"])

Durable outputs are the fourth layer. Write the object first, then register its URI; Rebase records the pointer and does not upload or proxy the bytes:

uri = write_to_object_storage(...)
artifact = rb.artifact(
    "Forecast parquet",
    uri=uri,
    key="forecast/SE3",
    media_type="application/parquet",
    metadata={"zone": "SE3"},
)

for artifact in run.artifacts():
    print(artifact["name"], artifact["uri"])

Artifacts created inside rb.task and Function.map calls inherit that task's identity automatically. Use disposition="reused" when code found the durable output instead of creating it. A hosted registration failure raises ArtifactReportingError; outside a hosted run the call is a validated in-memory no-op.

See Tasks and Artifacts for complete examples, including ThreadPoolExecutor context propagation and Function.map fan-out.

Attribution happens automatically inside deployed code. The runner injects the run and step IDs into the environment, and rb.current_run() exposes them:

ctx = rb.current_run()
if ctx is not None:
    print(ctx.run_id, ctx.step_run_id)   # step_run_id is None outside a step

current_run() returns None when the code is not running on the platform — a Function.map from a laptop belongs to no run, and the platform accepts the unattributed batch for exactly that case.

Use the CLI when you want to inspect a run from the terminal:

rebase run logs <run-id>

Standalone Registration

@rb.workflow(...) and @rb.function(...) work when you want a top-level handle without creating a project object first:

@rb.workflow(name="daily-forecast")
def forecast(site_id: str = "site-001") -> dict:
    return {"site_id": site_id}

forecast.deploy()
result = forecast.remote(site_id="site-001")

Prefer rb.project(...).workflow(...) when a file contains multiple functions and workflows that should deploy together.

HTTP errors are raised as RebaseWorkflowError with the response body attached.