Migrate from Baseten

​

Concept Mapping

​

Choosing a Path

​

Migration Checklist

1. Create one Python file next to your Truss. Everything in config.yaml and model/model.py ends up here.
2. Convert the class. class Model: becomes class MyApp(fal.App), load(self) becomes setup(self), and predict(self, model_input) becomes a method decorated with @fal.endpoint("/"). Drop __init__ unless it does work beyond capturing kwargs.
3. Move config.yaml keys to class attributes using the Concept Mapping table. requirements carries over verbatim. resources.accelerator becomes machine_type via the GPU Mapping table.
4. Type the interface. Replace the untyped model_input dict with a Pydantic Input model, and return a Pydantic Output model. Replace base64-encoded files with fal.toolkit types such as Image or File.
5. Handle python_version. If your Truss pinned it and your dependency pins depend on it, add a pyproject.toml. See Python Version.
6. Recreate secrets. For each key under secrets: in config.yaml, run fal secrets set NAME=value, and replace self._secrets["name"] with os.getenv("NAME").
7. Delete the cache configuration. model_cache and external_data have no equivalent and are not needed. See Model Weights and Caching.
8. Validate with fal run <file>::<Class> before deploying. This runs setup() on a real runner. If you added a pyproject.toml in step 5, the argument is the app key instead: fal run <app-key>.
9. Deploy with fal deploy <file>::<Class>, or fal deploy <app-key> when using pyproject.toml.
10. Repoint callers from https://model-{MODEL_ID}.api.baseten.co/... to fal_client.subscribe("your-username/your-app"). See Calling Your App.
11. Re-baseline output tests. Seeded outputs shift when the GPU changes. See the note under GPU Mapping.

​

Migration Path 1: A Truss Model

model_name: sdxl-turbo
python_version: py311
resources:
  accelerator: T4
  use_gpu: true
  cpu: "4"
  memory: 16Gi
requirements:
  - torch==2.4.0
  - diffusers==0.30.3
  - transformers==4.44.2
  - accelerate==0.34.2
  - safetensors==0.4.5
  - pillow==10.4.0
runtime:
  predict_concurrency: 1
secrets:
  hf_access_token: null

import base64
import io


class Model:
    def __init__(self, **kwargs):
        self._secrets = kwargs.get("secrets")
        self.pipe = None

    def load(self):
        import torch
        from diffusers import AutoPipelineForText2Image

        self.pipe = AutoPipelineForText2Image.from_pretrained(
            "stabilityai/sdxl-turbo",
            torch_dtype=torch.float16,
            variant="fp16",
        ).to("cuda")

    def predict(self, model_input):
        import torch

        prompt = model_input["prompt"]
        seed = int(model_input.get("seed", 42))
        steps = int(model_input.get("num_inference_steps", 4))

        generator = torch.Generator(device="cuda").manual_seed(seed)
        image = self.pipe(
            prompt=prompt,
            num_inference_steps=steps,
            guidance_scale=0.0,
            generator=generator,
        ).images[0]

        buffer = io.BytesIO()
        image.save(buffer, format="PNG")
        return {
            "image_b64": base64.b64encode(buffer.getvalue()).decode("utf-8"),
            "seed": seed,
            "prompt": prompt,
        }

import fal
from fal.toolkit import Image
from pydantic import BaseModel, Field


class Input(BaseModel):
    prompt: str = Field(description="The prompt to generate an image from.")
    seed: int = Field(default=42, description="Seed for reproducible generation.")
    num_inference_steps: int = Field(default=4, ge=1, le=10)


class Output(BaseModel):
    image: Image
    seed: int
    prompt: str


class SDXLTurbo(fal.App, name="sdxl-turbo"):
    machine_type = "GPU-A100"
    requirements = [
        "torch==2.4.0",
        "diffusers==0.30.3",
        "transformers==4.44.2",
        "accelerate==0.34.2",
        "safetensors==0.4.5",
        "pillow==10.4.0",
    ]

    def setup(self):
        import torch
        from diffusers import AutoPipelineForText2Image

        self.pipe = AutoPipelineForText2Image.from_pretrained(
            "stabilityai/sdxl-turbo",
            torch_dtype=torch.float16,
            variant="fp16",
        ).to("cuda")

    @fal.endpoint("/")
    def generate(self, input: Input) -> Output:
        import torch

        generator = torch.Generator(device="cuda").manual_seed(input.seed)
        image = self.pipe(
            prompt=input.prompt,
            num_inference_steps=input.num_inference_steps,
            guidance_scale=0.0,
            generator=generator,
        ).images[0]

        return Output(
            image=Image.from_pil(image),
            seed=input.seed,
            prompt=input.prompt,
        )

• config.yaml disappeared. resources.accelerator became machine_type and requirements became a class attribute. python_version is the one field with no class-attribute equivalent, and it is the first thing most Truss migrations trip over. See Python Version.
• load became setup. Same contract: it runs once per runner, and whatever you hang off self survives for that runner’s lifetime.
• predict became an endpoint. @fal.endpoint("/") mounts the handler at the app root, which is what fal_client calls by default.
• The untyped model_input dict became a Pydantic model. Baseten hands you whatever JSON arrived. fal validates against Input first and rejects malformed requests before your code runs. The same models generate your app’s OpenAPI spec, which powers the Playground and the generated client code.
• Base64 became a hosted file. fal.toolkit.Image.from_pil() uploads the image to fal’s CDN and returns a URL in the response, so large outputs never travel through your JSON payload. See Working with Files.
• predict_concurrency: 1 is the fal default. A runner handles one request at a time unless you raise max_multiplexing.
• The declared secret went away. hf_access_token was never read by this model. If yours does read one, see Secrets below.

​

Deploying

# Baseten
truss push --promote

# fal
fal deploy sdxl_turbo.py::SDXLTurbo

fal run sdxl_turbo.py::SDXLTurbo

​

Calling Your App

import os
import requests

model_id = "YOUR_MODEL_ID"

response = requests.post(
    f"https://model-{model_id}.api.baseten.co/environments/production/predict",
    headers={"Authorization": f"Bearer {os.environ['BASETEN_API_KEY']}"},
    json={"prompt": "an origami fox in a snowy forest", "seed": 42},
)
print(response.json())

import fal_client

result = fal_client.subscribe(
    "your-username/sdxl-turbo",
    arguments={"prompt": "an origami fox in a snowy forest", "seed": 42},
)
print(result["image"]["url"])

​

Migration Path 2: Custom Servers

base_image:
  image: your-registry/your-image:latest
docker_server:
  start_command: your-server --host 0.0.0.0 --port 8000
  server_port: 8000
  predict_endpoint: /predict
  readiness_endpoint: /health
  liveness_endpoint: /health

[tool.fal.apps.my-server]
auth = "private"
machine_type = "GPU-H100"
exposed_port = 8000
keep_alive = 300

[tool.fal.apps.my-server.image]
image = "your-registry/your-image:latest"
cmd = ["your-server", "--host", "0.0.0.0", "--port", "8000"]

COPY services/stats/server.py /opt/srv/server.py

import fal
from fal.container import ContainerImage

class MyApp(fal.App):
    machine_type = "GPU-H100"
    image = ContainerImage.from_dockerfile_str("""
        FROM pytorch/pytorch:2.4.0-cuda12.1-cudnn9-runtime
        RUN apt-get update && apt-get install -y ffmpeg && rm -rf /var/lib/apt/lists/*
        RUN pip install --no-cache-dir diffusers transformers accelerate
    """)

​

Migration Path 3: Chains

import truss_chains as chains

class Transcribe(chains.ChainletBase):
    remote_config = chains.RemoteConfig(
        docker_image=chains.DockerImage(pip_requirements=["whisper"]),
        compute=chains.Compute(gpu="L4"),
    )

    async def run_remote(self, audio_url: str) -> str:
        return transcribe(audio_url)


@chains.mark_entrypoint
class Summarize(chains.ChainletBase):
    remote_config = chains.RemoteConfig(compute=chains.Compute(gpu="H100"))

    def __init__(self, transcribe=chains.depends(Transcribe)):
        self._transcribe = transcribe

    async def run_remote(self, audio_url: str) -> str:
        text = await self._transcribe.run_remote(audio_url)
        return summarize(text)

import fal
import fal_client
from pydantic import BaseModel


class AudioInput(BaseModel):
    audio_url: str


class Transcript(BaseModel):
    text: str


class Summary(BaseModel):
    summary: str


class Transcribe(fal.App, name="transcribe"):
    machine_type = "GPU-A100"
    requirements = ["whisper"]

    @fal.endpoint("/")
    def run(self, input: AudioInput) -> Transcript:
        return Transcript(text=transcribe(input.audio_url))


class Summarize(fal.App, name="summarize"):
    machine_type = "GPU-H100"

    @fal.endpoint("/")
    def run(self, input: AudioInput) -> Summary:
        transcript = fal_client.subscribe(
            "your-username/transcribe",
            arguments={"audio_url": input.audio_url},
        )
        return Summary(summary=summarize(transcript["text"]))

​

Key Differences

​

Python Version

× No solution found when resolving dependencies:
╰─▶ Because torch==2.4.0 has no wheels with a matching Python ABI tag
    (e.g., `cp314`) and you require torch==2.4.0, we can conclude that your
    requirements are unsatisfiable.
    hint: You require CPython 3.14 (`cp314`), but we only found wheels for
    `torch` (v2.4.0) with the following Python ABI tags: `cp38`, `cp39`,
    `cp310`, `cp311`, `cp312`

[tool.fal.apps.sdxl-turbo]
ref = "sdxl_turbo.py::SDXLTurbo"
auth = "private"
python_version = "3.11"
machine_type = "GPU-A100"

fal deploy sdxl-turbo

✘ Local Python 3.14 differs from the app's python_version=3.11.
  Run from a Python 3.11 interpreter, or set python_version='3.14' on the app.

​

GPU Mapping

import fal

class MyApp(fal.App):
    machine_type = "GPU-H100"
    num_gpus = 8

import fal

class MyApp(fal.App):
    machine_type = ["GPU-H100", "GPU-A100"]

​

Autoscaling

import fal

class MyApp(fal.App):
    machine_type = "GPU-H100"
    keep_alive = 300
    min_concurrency = 1
    max_concurrency = 10
    max_multiplexing = 1

fal apps scale my-app --min-concurrency 2 --max-concurrency 20 --keep-alive 600

​

Queue-Based Inference

resp = requests.post(
    f"https://model-{model_id}.api.baseten.co/production/async_predict",
    headers={"Authorization": f"Bearer {api_key}"},
    json={
        "model_input": {"prompt": "hello world"},
        "webhook_endpoint": "https://your-server.com/webhook",
        "priority": 0,
    },
)
request_id = resp.json()["request_id"]

import fal_client

handler = fal_client.submit(
    "your-username/sdxl-turbo",
    arguments={"prompt": "hello world"},
    webhook_url="https://your-server.com/webhook",
)
request_id = handler.request_id

• No separate route. subscribe and submit both go through the queue; subscribe just polls for you. Baseten needs you to choose /predict or /async_predict up front.
• The input is not wrapped. Baseten nests your payload under model_input; fal sends it at the top level.
• Status polling looks the same. handler.status() and handler.get() replace GET /async_request/{request_id}.

​

Secrets

secrets:
  hf_access_token: null

class Model:
    def __init__(self, **kwargs):
        self._secrets = kwargs["secrets"]

    def load(self):
        import huggingface_hub
        huggingface_hub.login(token=self._secrets["hf_access_token"])

fal secrets set HF_TOKEN=hf_abc123

import os
import fal

class MyApp(fal.App):
    def setup(self):
        import huggingface_hub
        huggingface_hub.login(token=os.getenv("HF_TOKEN"))

import fal

class MyApp(fal.App):
    secrets = ["HF_TOKEN"]

​

Model Weights and Caching

import os
import fal

class MyApp(fal.App):
    def setup(self):
        weights = "/data/models/my-model.safetensors"
        if not os.path.exists(weights):
            download_to(weights)
        self.model = load(weights)

​

Preprocess and Postprocess

• Do the I/O inline. Raise max_multiplexing so a runner can overlap the I/O of one request with the GPU work of another. Your handlers must be safe to run concurrently.
• Split the app. A CPU app fetches and normalizes inputs, then calls the GPU app. This is Multi-App Routing, and it keeps the expensive machine doing only GPU work.

​

Streaming and WebSockets

import fal
from fastapi.responses import StreamingResponse

class MyApp(fal.App):
    @fal.endpoint("/")
    def generate(self, input: Input) -> Output:
        ...

    @fal.endpoint("/stream")
    def generate_stream(self, input: Input) -> StreamingResponse:
        ...

​

Deployments and Environments

fal environments create staging --description "Staging environment"
fal deploy sdxl_turbo.py::SDXLTurbo --env staging

​

CLI Mapping

​

Troubleshooting

​

Next Steps