Real-time endpoints & WebSockets
For applications deployed on fal runtime; in addition to regular HTTP endpoints, developers might choose to implement auxiliary interfaces on top of raw WebSockets or fal's (stateless) real-time application framework.
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 (opens in a new tab). 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.
import fal
from pydantic import BaseModel
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.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()
@fal.realtime("/realtime")
def generate_rt(self, input: Input) -> Output:
return Output(output=input.prompt)