ASGI, WSGI, Uvicorn and Gunicorn

What we're doing: Run both a WSGI app and an ASGI app by simulating exactly what the server side of each contract does, without opening a real socket.

import asyncio

def wsgi_app(environ, start_response):
    start_response("200 OK", [("Content-Type", "text/plain")])
    return [f"Hello, {environ['PATH_INFO']}".encode()]

captured = {}
def fake_start_response(status, headers, exc_info=None):
    captured["status"] = status

result = wsgi_app({"PATH_INFO": "/world"}, fake_start_response)
print("WSGI:", captured["status"], b"".join(result).decode())


async def asgi_app(scope, receive, send):
    await receive()
    await send({"type": "http.response.start", "status": 200,
                 "headers": [(b"content-type", b"text/plain")]})
    await send({"type": "http.response.body",
                 "body": f"Hello, {scope['path']}".encode()})

async def run_asgi():
    sent = []
    async def receive():
        return {"type": "http.request", "body": b""}
    async def send(message):
        sent.append(message)
    await asgi_app({"type": "http", "path": "/world"}, receive, send)
    return sent

for msg in asyncio.run(run_asgi()):
    print("ASGI:", msg)

3

wsgi_app takes environ (a dict of request data) and start_response (a callback to set status/headers) — one synchronous call, one return value.

15

asgi_app takes scope, receive, and send — three parameters instead of two, and the function itself is async.

17

The app awaits receive() to get the next event instead of it being handed in as an argument — this is what lets it wait for more data without blocking the whole process.

19

The response goes out as two separate send() calls (start, then body) instead of one return value — this is what lets ASGI stream a response.

WSGI: 200 OK Hello, /world
ASGI: {'type': 'http.response.start', 'status': 200, 'headers': [(b'content-type', b'text/plain')]}
ASGI: {'type': 'http.response.body', 'body': b'Hello, /world'}

Why this works: Both callables do the same job — read the request, write a 200 with a text body — but WSGI does it as one function call that returns everything at once, while ASGI does it as a sequence of awaited messages. That difference is the entire reason ASGI exists: a WSGI worker cannot do anything else while wsgi_app runs, but an ASGI app can await receive() and let the event loop serve other connections while it waits.

What we're doing: Configure Uvicorn programmatically with uvicorn.Config and confirm the real kwargs it accepts for host, port, worker count, and graceful shutdown timeout.

import uvicorn

cfg = uvicorn.Config(
    "main:app",
    host="0.0.0.0",
    port=8000,
    workers=4,
    timeout_graceful_shutdown=30,
    proxy_headers=True,
)
print(cfg.host, cfg.port, cfg.workers, cfg.timeout_graceful_shutdown)

3

"main:app" is module:attribute — Uvicorn imports main.py and looks up the app object inside it.

6

workers=4 is only meaningful when running via the CLI or uvicorn.run — Config itself just stores the value.

7

timeout_graceful_shutdown caps how long Uvicorn waits for in-flight requests to finish before force-killing on shutdown.

0.0.0.0 8000 4 30

Why this works: uvicorn.Config accepts these exact keyword names — they map one-to-one onto the CLI flags (--host, --port, --workers, --timeout-graceful-shutdown), so a config object and a CLI invocation are two ways of setting the same underlying values.

What we're doing: Read the exact Gunicorn CLI flags and defaults for worker count, worker class, threads, and graceful shutdown timeout, as documented.

# -w / --workers        default: 1
# -k / --worker-class    default: 'sync'   (choices: sync, gthread, gevent, tornado, or a class path)
# --threads              default: 1        (only used by the gthread worker class)
# -t / --timeout         default: 30       (seconds a worker may be silent before Gunicorn kills+restarts it)
# --graceful-timeout     default: 30       (seconds a worker gets to finish in-flight work after a restart/shutdown signal)
# -b / --bind            default: '127.0.0.1:8000'
# --preload              default: False    (load app code once before forking workers)

gunicorn -w 4 -k uvicorn.workers.UvicornWorker --graceful-timeout 30 --bind 0.0.0.0:8000 main:app

2

Gunicorn's own default worker count is 1 — production deployments always override this explicitly, commonly to (2 * CPU cores) + 1.

3

worker-class defaults to 'sync': one request at a time per worker process, no threading.

5

timeout is a liveness check — a worker that goes silent (e.g. stuck on a slow synchronous call) past this many seconds gets killed and replaced.

6

graceful-timeout is different: it applies during a deliberate restart/shutdown, giving in-flight requests time to finish instead of being cut off immediately.

Why this works: Every flag here maps to a documented Gunicorn setting with a stated default — none of it is Uvicorn's job. Gunicorn owns process count and lifecycle; the worker class (sync, gthread, or Uvicorn's ASGI worker) owns how one process actually serves requests.

What we're doing: Show why an async worker (one event loop) can hold many concurrent connections without extra OS threads, using the stdlib event loop directly.

import asyncio
import time

async def handle_request(conn_id, delay):
    print(f"conn {conn_id}: waiting on I/O")
    await asyncio.sleep(delay)          # simulates an awaited DB/HTTP call
    print(f"conn {conn_id}: done")
    return conn_id

async def main():
    start = time.perf_counter()
    # One event loop, three "connections" in flight concurrently —
    # no extra threads or processes were created for this.
    results = await asyncio.gather(
        handle_request(1, 0.2),
        handle_request(2, 0.2),
        handle_request(3, 0.2),
    )
    elapsed = time.perf_counter() - start
    print(f"handled {results} in {elapsed:.2f}s")

asyncio.run(main())

4

handle_request represents one connection's work — the kind of function an ASGI app runs per request.

6

await asyncio.sleep(delay) simulates waiting on I/O (a DB query, an HTTP call) — this is where the event loop switches to another connection instead of blocking.

12

asyncio.gather runs all three concurrently on ONE event loop — no threads, no extra processes — the same mechanism an async worker uses across many real connections.

conn 1: waiting on I/O
conn 2: waiting on I/O
conn 3: waiting on I/O
conn 1: done
conn 2: done
conn 3: done
handled [1, 2, 3] in 0.20s

Why this works: All three 0.2s waits overlap — total time is ~0.2s, not 0.6s — because the event loop starts connection 2's work the instant connection 1 hits an await, rather than waiting for connection 1 to fully finish. That is exactly the mechanism an async worker (Uvicorn, or Gunicorn + UvicornWorker) uses to serve many concurrent requests from a single OS process: no thread is blocked, so nothing needs to be created per connection.

What we're doing: Uvicorn only trusts X-Forwarded-* headers from an explicitly allowed proxy address — confirm the real config keys that control this.

import uvicorn

cfg = uvicorn.Config(
    "main:app",
    proxy_headers=True,
    forwarded_allow_ips="10.0.0.5",   # only trust this proxy's IP
)
print(cfg.proxy_headers, cfg.forwarded_allow_ips)

5

proxy_headers=True tells Uvicorn to read X-Forwarded-For/X-Forwarded-Proto at all — off by default would mean the app sees the proxy's own IP/scheme, not the real client's.

6

forwarded_allow_ips restricts which upstream IP is trusted to set those headers — without this, anyone who can reach the app directly could spoof their own IP.

Why this works: Trusting forwarded headers from every source would let a client bypass IP-based rate limiting or logging simply by setting X-Forwarded-For itself. Scoping trust to the known proxy address is what makes the headers a safe source of the real client's info.

What we're doing: Drive the raw ASGI lifespan protocol end to end — startup event in, startup-complete out; shutdown event in, shutdown-complete out — the mechanism FastAPI's lifespan= wraps.

import asyncio

resources = {}

async def app(scope, receive, send):
    if scope["type"] != "lifespan":
        return
    while True:
        message = await receive()
        if message["type"] == "lifespan.startup":
            resources["db_pool"] = "connected"
            print("startup: opened", resources["db_pool"])
            await send({"type": "lifespan.startup.complete"})
        elif message["type"] == "lifespan.shutdown":
            print("shutdown: closing", resources["db_pool"])
            resources.clear()
            await send({"type": "lifespan.shutdown.complete"})
            return

async def simulate_server():
    events = [{"type": "lifespan.startup"}, {"type": "lifespan.shutdown"}]
    sent = []
    async def receive():
        return events.pop(0)
    async def send(message):
        sent.append(message)
    await app({"type": "lifespan"}, receive, send)
    return sent

for msg in asyncio.run(simulate_server()):
    print("server saw:", msg)

5

The app checks scope["type"] == "lifespan" to handle this specially — it is a different scope from a normal HTTP request.

9

lifespan.startup arrives exactly once, before the server routes any HTTP requests to this app — the correct place to open a DB pool.

14

lifespan.shutdown arrives once, after the server has stopped accepting new connections and drained in-flight ones — the correct place to close what startup opened.

startup: opened connected
server saw: {'type': 'lifespan.startup.complete'}
shutdown: closing connected
server saw: {'type': 'lifespan.shutdown.complete'}

Why this works: This is the exact protocol Uvicorn speaks to any ASGI app at process start and stop, and what a framework's lifespan()/on_event("startup") wraps in a friendlier API. Startup runs once, before the first real request; shutdown runs once, only after the server has already stopped taking new connections and drained the in-flight ones — so a DB pool closed here is not closed out from under a request still being served.