12  ASGI, Starlette, FastAPI

What WSGI Can’t Do

Every WSGI request is a synchronous Python call to the application() interface we’ve seen.

This works fine for short-lived requests, but breaks down when:

  • A request takes a long time (slow DB, external API)
  • You need to push data to the client continuously (WebSockets)
import time
from wsgiref.simple_server import make_server

def app(environ, start_response):
    time.sleep(3)   # database read or upstream API call
    start_response("200 OK", [("Content-Type", "text/plain")])
    return [b"done\n"]

with make_server("", 8000, app) as server:
    server.serve_forever()

Run a few requests simultaneously, and we see that they queue– only one is being handled at a time, causing the third request to wait 9 seconds. In reality these would quickly start to time out.

"""
Make N simultaneous requests and time them.

In practice, these requests could be coming from different users
but we'll use async to simulate many users from one machine.
"""

import asyncio
import httpx
import sys
import time

URL = "http://localhost:8000"
N = int(sys.argv[1]) if len(sys.argv) > 1 else 3


async def fetch(client, i):
    start = time.perf_counter()
    await client.get(URL)
    elapsed = time.perf_counter() - start
    print(f"request {i:2d}: {elapsed:.1f}s")


async def main():
    async with httpx.AsyncClient(timeout=None) as client:
        start = time.perf_counter()
        print(f"starting {N} simultaneous requests...")
        await asyncio.gather(*[fetch(client, i + 1) for i in range(N)])
        print(f"total: {time.perf_counter() - start:.1f}s")


asyncio.run(main())

ASGI

The WSGI contract assumes one request, one response. Async web applications need to keep a bidirectional connection open so they can continue to send information in both directions.

Asynchronous Server Gateway Interface, https://asgi.readthedocs.io/en/latest/

# WSGI
def app(environ, start_response): ...
# ASGI
async def app(scope, receive, send): ...
  • scope - request metadata (type, path, headers) - similar to WSGI environ
  • receive - async function to read incoming events
  • send - async function to push response chunks
import asyncio

async def app(scope, receive, send):
    # scope replaces environ
    assert scope["type"] == "http"

    # similar delay
    await asyncio.sleep(3)

    # send() with different message types replaces the
    # start_response/return pattern from WSGI
    await send({
        "type": "http.response.start",
        "status": 200,
        "headers": [[b"content-type", b"text/plain"]],
    })
    await send({
        "type": "http.response.body",
        "body": b"hello from ASGI\n",
    })
uv run uvicorn 01asgi:app

WebSockets

HTTP is request/response: the client asks, the server answers, the connection closes.

WebSockets keep the connection open, allowing both sides to send messages at any time after established.

WebSocket handshake:

  1. Client sends an HTTP request with Upgrade: websocket
  2. Server responds with 101 Switching Protocols
  3. Connection stays open, now a persistent channel:
ws://localhost:8000/chat     # unencrypted
wss://example.com/chat       # encrypted (like https)
async def app(scope, receive, send):
    if scope["type"] == "websocket":
        # establish connection
        event = await receive() # wait for incoming event["type"] == "websocket.connect"
        await send({"type": "websocket.accept"})

        # long-lived async coroutine
        while True:
            event = await receive()
            if event["type"] == "websocket.disconnect":
                break
            await send({
                "type": "websocket.send",
                "text": f"echo: {event['text']}"
            })
// Browser JS Client
const ws = new WebSocket("ws://localhost:8000/ws");
ws.onmessage = (event) => console.log(event.data);
ws.send("hello");

Starlette

Starlette is the Flask-equivalent for ASGI: a thin layer adding routing, request/response objects, WebSocket handling.

from starlette.applications import Starlette
from starlette.requests import Request
from starlette.responses import Response, JSONResponse
from starlette.routing import Route, WebSocketRoute
from starlette.websockets import WebSocket


# standard request -> response views, main difference: can use await within
async def homepage(request: Request) -> Response:
    return JSONResponse({"hello": "world"})


# websocket views: again, long running coroutines 
async def ws(websocket: WebSocket):
    await websocket.accept()
    while True:
        text = await websocket.receive_text()
        await websocket.send_text(f"echo: {text}")


# base-level ASGI app, similar to Flask
app = Starlette(
    # explicitly build routing table, similar to Django
    routes=[
        Route("/", homepage),
        WebSocketRoute("/ws", ws),
    ]
)

Key additions over raw ASGI:

  • Route, WebSocketRoute - declarative routing
  • Request - wraps scope/receive into a familiar interface
  • JSONResponse, HTMLResponse - convenience response types
  • WebSocket - connection object with .accept(), .send_text(), .receive_text()

FastAPI

FastAPI is built on top of Starlette, and mainly adds two things on top of Starlette that are useful to almost any API:

  • Pydantic type annotations become request/response validation
  • OpenAPI automatic interactive docs at /docs

Sticky Notes (Async)

import random
from fastapi import FastAPI
from fastapi.responses import HTMLResponse
from pydantic import BaseModel
from datetime import datetime

app = FastAPI()

COLORS = ["khaki", "lightblue", "lightgreen", "lightsalmon", "plum"]


class NoteIn(BaseModel):
    text: str


class Note(BaseModel):
    text: str
    color: str
    created_at: datetime


notes: list[Note] = []


@app.get("/", response_class=HTMLResponse)
def index():
    with open("05notes.html") as f:
        return f.read()


@app.get("/api/notes")
def get_notes():
    return notes


@app.post("/api/notes", status_code=201)
def add_note(note: NoteIn):
    notes.append(
        Note(text=note.text, color=random.choice(COLORS), created_at=datetime.now())
    )
    return notes

uv run uvicorn 05notes:app

http://localhost:8000/docs - API is fully documented and testable with no extra work.

The client-side code is identical to the Flask version. The difference is the server can now handle concurrent requests without blocking.

<input id="t" placeholder="note...">
<button onclick="post()">Add</button>
<div id="wall"></div>

<script>
function render(notes) {
  document.getElementById("wall").innerHTML = notes.map(n =>
    `<div class="note" style="background:${n.color}">
      ${n.text}
      <small>${new Date(n.created_at).toLocaleTimeString()}</small>
    </div>`
  ).join("");
}

async function post() {
  const res = await fetch("/api/notes", {
    method: "POST",
    headers: {"Content-Type": "application/json"},
    body: JSON.stringify({text: document.getElementById("t").value})
  });
  render(await res.json());
}

fetch("/api/notes").then(r => r.json()).then(render);
</script>

<style>
.note { padding: 1em; margin: .5em; display: inline-block; width: 8em;
        vertical-align: top; word-wrap: break-word; }
.note small { display: block; font-size: 0.7em; margin-top: 0.5em; opacity: 0.6; }
</style>

This delivers async updates to the server, but clients are not automatically informed when a new post is made. This is sometimes fine! We can handle this with polling, having the client ask periodically–“any new mail?”.

If we want clients to be notified automatically, we need to stay on the line, that’s where web sockets come in:

Sticky Notes (Web Sockets)

import random
from fastapi import FastAPI, WebSocket, WebSocketDisconnect
from fastapi.responses import HTMLResponse
from pydantic import BaseModel
from datetime import datetime

app = FastAPI()
COLORS = ["khaki", "lightblue", "lightgreen", "lightsalmon", "plum"]


class NoteIn(BaseModel):
    text: str


class Note(BaseModel):
    text: str
    color: str
    created_at: datetime


notes: list[Note] = []


class ConnectionManager:
    def __init__(self):
        self.connections: list[WebSocket] = []

    async def connect(self, ws: WebSocket):
        await ws.accept()
        self.connections.append(ws)

    def disconnect(self, ws: WebSocket):
        self.connections.remove(ws)

    async def broadcast(self, message: str):
        for ws in self.connections:
            await ws.send_text(message)


manager = ConnectionManager()


@app.get("/", response_class=HTMLResponse)
def index():
    with open("06notes_ws.html") as f:
        return f.read()


@app.get("/api/notes")
def get_notes():
    return notes


@app.post("/api/notes", status_code=201)
async def add_note(note: NoteIn):
    n = Note(text=note.text, color=random.choice(COLORS), created_at=datetime.now())
    notes.append(n)
    # NEW: send update to all connected clients
    await manager.broadcast(n.model_dump_json())
    return notes


@app.websocket("/ws")
async def websocket_endpoint(ws: WebSocket):
    # create a connection and receive messages
    await manager.connect(ws)
    try:
        while True:
            await ws.receive_text()  # keep connection alive
    except WebSocketDisconnect:
        manager.disconnect(ws)
<input id="t" placeholder="note...">
<button onclick="post()">Add</button>
<div id="wall"></div>

<script>
const ws = new WebSocket("ws://localhost:8000/ws");

ws.onmessage = (event) => {
  const n = JSON.parse(event.data);
  addNote(n);
};

function addNote(n) {
  const div = document.createElement("div");
  div.className = "note";
  div.style.background = n.color;
  div.innerHTML = `${n.text}<small>${new Date(n.created_at).toLocaleTimeString()}</small>`;
  document.getElementById("wall").prepend(div);
}

async function post() {
  await fetch("/api/notes", {
    method: "POST",
    headers: {"Content-Type": "application/json"},
    body: JSON.stringify({text: document.getElementById("t").value})
  });
  document.getElementById("t").value = "";
}

fetch("/api/notes").then(r => r.json()).then(notes => notes.forEach(addNote));
</script>

<style>
.note { padding: 1em; margin: .5em; display: inline-block; width: 8em;
        vertical-align: top; word-wrap: break-word; }
.note small { display: block; font-size: 0.7em; margin-top: 0.5em; opacity: 0.6; }
</style>

The ConnectionManager maintains a list of active WebSocket connections and broadcasts each message to all of them. Open three browser tabs — they all receive every message.

Demo

uv run uvicorn 06chat:app --reload

Open http://localhost:8000 in multiple tabs. Each tab gets a separate WebSocket connection. Updates sent from one tab appear in all others instantly.

This is the logic behind chat apps, collaborative editors, live dashboards, and API streaming responses such as those from LLMs.