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

Steaming Responses

You can setup streaming responses using the FastAPI StreamingResponse. Here’s an example:

import time

from beam import App, Runtime, Image
from fastapi import FastAPI
from fastapi.responses import StreamingResponse, JSONResponse


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


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

    @api.get("/")
    def stream_html():
        def _stream_text():
            message = "Hello, this is a streaming response from Beam!"
            for character in message:
                # Stream each character
                yield character
                time.sleep(0.1)  # Simulate delay for typing effect

            # End with the closing HTML tags
            yield "</p></body></html>"

        return StreamingResponse(_stream_text(), media_type="text/html")
    
    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=1, authorized=False)
def web_server():
    return my_app

Invoking a websocket

After you’ve deployed your code by running beam deploy [app.py], it’s time to invoke it.

For a quick and dirty way of testing websockets, you can use websocat. If you don’t have websocat installed, you can do so with brew install websocat.

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

Here’s how to invoke the websocket app in Python and Javascript:

When you run this code, you’ll see the websocket echo back to your shell:

(.venv) user@MacBook websocket-app % python request.py
Connection established.
Received from server: Message text was: 0
Received from server: Message text was: 1
Received from server: Message text was: 2
Received from server: Message text was: 3

Response Types

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