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:

import asyncio
import websockets

async def connect():
    uri = "wss://[APP-ID].apps.beam.cloud"  # Adjust as necessary
    async with websockets.connect(uri) as ws:
        print("Connection established.")

        # Send a message when connection is established
        await ws.send("Hello Server!")

        # Keep connection open
        while True:
            # Wait for message from the server
            message = await ws.recv()
            print(f"Received from server: {message}")
            
            # Send a response back to the server
            await ws.send(f"Echo: {message}")

if __name__ == "__main__":
    asyncio.run(connect())

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.

Was this page helpful?

Scheduled JobDeployment Previews