Skip to main content
Under a fal.App, for any endpoint that deal with real-time connectivity, @fal.realtime() decorator can be used instead of @fal.endpoint to automatically make the interface compatible with fal’s real-time clients. The functions do not provide any session state, and are meant to be used for reducing the overall latency (with fal’s binary protocol) and eliminating fixed connection establishing overheads. For power users who want to build stateful applications with their own real-time protocol, a @fal.endpoint can be initialized with is_websocket=True flag and the underlying function will receive the raw WebSocket connection and can choose to use it however it wants. See examples of both approaches below: 
import fal
from pydantic import BaseModel
from fastapi import WebSocket


class Input(BaseModel):
    prompt: str = Field()


class Output(BaseModel):
    output: str = Field()


class RealtimeApp(fal.App):
    @fal.endpoint("/")
    def generate(self, input: Input) -> Output:
        return Output(output=input.prompt)

    @fal.realtime("/realtime")
    def generate_rt(self, input: Input) -> Output:
        return Output(output=input.prompt)

    @fal.endpoint("/ws", is_websocket=True)
    async def generate_ws(self, websocket: WebSocket) -> None:
        await websocket.accept()
        for _ in range(3):
            await websocket.send_json({"message": "Hello world!"})
        await websocket.close()