You can run ASGI-compatible apps on Beam, including any existing FastAPI or Starlette app.

Here’s a basic example of a FastAPI app. It can be invoked with a GET request, and it returns {"hello": "world"}

app.py
from beam import App, Runtime, Image
from fastapi import FastAPI

app = App(
    name="asgi-app",
    runtime=Runtime(image=Image(python_packages=["fastapi"])),
)


@app.asgi(authorized=False)
def web_server():
    api = FastAPI()

    @api.get("/")
    def hello_world():
        return {"hello": "world"}

    return api

You can deploy this code using beam deploy app.py, and navigate to the URL in your browser:

https://[APP-ID].apps.beam.cloud

{"hello":"world"}

​
HTTP Endpoints

Authorization can be toggled on or off through the authorized=False argument. If enabled, you will need to include your Beam auth token in the request.

​
Adding multiple API routes per app

You can use ASGI apps along with FastAPI to add multiple API routes to a single deployment. For example, you might want an app with two separate routes: one to train a model, and one for running inference.

Endpoint 1

https://[APP-ID].apps.beam.cloud/divide?x=10

Endpoint 2

https://[APP-ID].apps.beam.cloud/square?x=100

from beam import App, Runtime, Image
from fastapi import FastAPI

app = App(
    name="square",
    runtime=Runtime(
        cpu=1,
        memory="1Gi",
        image=Image(
            python_packages=["fastapi"]
        )
    ),
)

api = FastAPI()


@app.asgi(authorized=False)
def handler():
    @api.get("/square")
    def square(x: int):
        return {"square": x**2}

    @api.get("/divide")
    def divide(x: int):
        return {"dividend": x / 2}

    return api

​
Websockets

You can also interact with a Beam app via a websocket:

import asyncio

from beam import App, Image, Runtime
from fastapi import FastAPI, WebSocket

app = App(
    "ws-server",
    runtime=Runtime(
        cpu=1, memory="1Gi", image=Image(python_packages=["fastapi", "httpx", "requests"])
    ),
)

my_app = FastAPI()


@my_app.websocket("/")
async def ws_endpoint(websocket: WebSocket):
    await websocket.accept()
    for i in range(50):
        await websocket.send_text(f"Message text was: {i}")
        await asyncio.sleep(0.5)


@app.asgi(workers=2, authorized=False)
def web_server():
    return my_app

You can test this in your shell by using websocat.

If you don’t have websocat installed, you can do so with brew install websocat.

websocat wss://[YOUR-APP-ID].apps.stage.beam.cloud/

​
Response Types

Beam supports various response types, including any FastAPI response type. You can find a list of FastAPI response types here.

Was this page helpful?

Scheduled JobDeployment Previews