Loading FASTAPI::COSMOS...
“FastAPI natively supports asynchronous streaming patterns. BackgroundTasks queues lightweight jobs (sending welcome emails, logging audit trails) to run after the HTTP response has been sent to the client. Server-Sent Events (SSE) stream AI LLM tokens via sse-starlette, while WebSockets enable full-duplex real-time communication.”
Streaming real-time SSE tokens, full-duplex WebSockets, and firing background tasks after sending HTTP responses.
from fastapi import FastAPI, BackgroundTasks
from sse_starlette.sse import EventSourceResponse
import asyncio
app = FastAPI()
# 1. Background Task (Runs after HTTP Response)
@app.post("/register")
async def register(email: str, bg: BackgroundTasks):
bg.add_task(send_welcome_email, email)
return {"message": "User registered, email will send in background"}
# 2. Server-Sent Events (Real-time Token Stream)
@app.get("/stream-tokens")
async def stream_tokens():
async def token_generator():
for word in ["Building", "high-performance", "async", "APIs", "with", "FastAPI"]:
await asyncio.sleep(0.1)
yield {"data": word}
return EventSourceResponse(token_generator())Client invokes endpoint: e.g. POST /signup
Endpoint registers background job: background_tasks.add_task(send_email, user.email)
FastAPI returns HTTP 201 Created to client immediately (<10ms)
Event loop executes send_email() asynchronously in the background
For SSE: yields EventSourceMessage chunks continuously over persistent HTTP stream
Streaming LLM tokens with Server-Sent Events reduces Time to First Token (TTFT) perceived latency by 90% compared to waiting for full generation.