La diferencia entre WSGI y ASGI no es solo una cuestión de “síncrono vs asíncrono”: es el contrato que tu aplicación firma con el servidor, y ese contrato determina qué tipos de conexiones puedes manejar, cómo escala bajo carga y qué tecnologías de tiempo real están disponibles o simplemente prohibidas.

WSGI (definido en PEP 3333) es un callable Python que recibe environ y start_response, procesa la request de forma síncrona, y devuelve un iterable con el body. Un worker WSGI maneja exactamente una request a la vez; cuando esa llamada bloquea esperando la base de datos, el worker está muerto para todo lo demás. Gunicorn compensa esto lanzando múltiples procesos worker (tipicamente 2 * CPUs + 1), así que el paralelismo es real pero tiene un coste: cada worker es un proceso Python completo con su propio heap y su propio GIL.

ASGI (PEP no oficial, especificación en el repositorio de asgiref) introduce un contrato diferente: la aplicación es un callable async que recibe scope, receive y send. El scope describe la conexión (HTTP, WebSocket, lifespan); receive y send son coroutines que permiten al servidor y a la aplicación intercambiar mensajes de forma incremental. Un worker ASGI con asyncio puede tener miles de conexiones en vuelo simultáneamente porque mientras espera I/O en una, el event loop avanza con otras.

Esto habilita cosas que WSGI hace imposibles por diseño: WebSockets necesitan una conexión persistente bidireccional; Server-Sent Events requieren que la respuesta quede abierta indefinidamente; HTTP/2 push necesita multiplexar streams sobre una sola conexión TCP. WSGI no tiene modelo mental para ninguno de esos casos porque su interfaz asume que la respuesta termina y el worker queda libre.

La pregunta relevante no es “¿cuál es mejor?” sino “¿cuándo el overhead de ASGI no se justifica?”. Si tu aplicación es mayoritariamente request/response sin I/O concurrente, Flask con Gunicorn es más predecible, tiene mejor ecosistema de middleware maduro y el modelo de procesos aislados te da fault isolation gratis. ASGI brilla cuando el I/O concurrente es real: muchas conexiones simultáneas a APIs externas, WebSockets, o cuando ya estás usando asyncio en toda la stack.

# wsgi_vs_asgi_demo.py
# Requiere: pip install flask fastapi uvicorn gunicorn httpx anyio

import asyncio
import time

import httpx
from flask import Flask
from fastapi import FastAPI
from fastapi.responses import StreamingResponse

# ── WSGI: Flask ──────────────────────────────────────────────────────────────

wsgi_app = Flask(__name__)

@wsgi_app.route("/wsgi/sequential")
def wsgi_sequential():
    """
    Cada llamada a time.sleep bloquea el worker completo.
    Con gunicorn -w 4, solo 4 requests pueden ejecutarse en paralelo.
    """
    time.sleep(0.1)   # simula I/O bloqueante (base de datos, API externa)
    time.sleep(0.1)
    return {"total_sleep": "200ms", "interface": "WSGI"}


# ── ASGI: FastAPI ────────────────────────────────────────────────────────────

asgi_app = FastAPI()

@asgi_app.get("/asgi/concurrent")
async def asgi_concurrent():
    """
    asyncio.sleep cede el control al event loop; el worker puede procesar
    otras requests mientras espera. Un solo proceso maneja miles de estas.
    """
    await asyncio.sleep(0.1)
    await asyncio.sleep(0.1)
    return {"total_sleep": "200ms", "interface": "ASGI"}


@asgi_app.get("/asgi/fan-out")
async def asgi_fan_out():
    """
    Fan-out a tres APIs externas en paralelo: imposible sin async en WSGI
    sin lanzar threads explícitamente.
    """
    async with httpx.AsyncClient() as client:
        results = await asyncio.gather(
            client.get("https://httpbin.org/delay/1"),
            client.get("https://httpbin.org/delay/1"),
            client.get("https://httpbin.org/delay/1"),
        )
    # Los tres requests salen al mismo tiempo; el tiempo total es ~1s, no ~3s.
    return {"statuses": [r.status_code for r in results]}


async def event_stream():
    """
    Server-Sent Events: la conexión permanece abierta y el servidor empuja
    mensajes. WSGI no puede modelar esto; ASGI lo hace de forma natural
    porque send/receive son streams, no un único par request/response.
    """
    for i in range(5):
        await asyncio.sleep(1)
        yield f"data: tick {i}\n\n"   # formato SSE estándar

@asgi_app.get("/asgi/sse")
async def asgi_sse():
    return StreamingResponse(event_stream(), media_type="text/event-stream")


# ── Inspección del contrato de cada interfaz ─────────────────────────────────

def inspect_wsgi_contract(app):
    """
    Un servidor WSGI llama a la aplicación así de simple.
    environ es un dict con los headers y metadatos HTTP.
    start_response es un callable que la app invoca para fijar status y headers.
    """
    responses = []

    def start_response(status, headers):
        responses.append({"status": status, "headers": headers})

    environ = {
        "REQUEST_METHOD": "GET",
        "PATH_INFO": "/wsgi/sequential",
        "SERVER_NAME": "localhost",
        "SERVER_PORT": "8000",
        "wsgi.input": __import__("io").BytesIO(b""),
        "wsgi.errors": __import__("sys").stderr,
        "wsgi.url_scheme": "http",
        "wsgi.multithread": False,
        "wsgi.multiprocess": True,
        "wsgi.run_once": False,
    }

    body = b"".join(app(environ, start_response))
    return responses[0]["status"], body


async def inspect_asgi_contract(app):
    """
    Un servidor ASGI llama a la aplicación con scope/receive/send.
    receive y send son coroutines: el protocolo es un diálogo, no un monólogo.
    """
    scope = {
        "type": "http",
        "asgi": {"version": "3.0"},
        "http_version": "1.1",
        "method": "GET",
        "path": "/asgi/concurrent",
        "query_string": b"",
        "headers": [],
    }

    messages = []

    async def receive():
        # El servidor entrega el body de la request cuando la app lo pide
        return {"type": "http.request", "body": b"", "more_body": False}

    async def send(message):
        # La app empuja mensajes al servidor: primero headers, luego body
        messages.append(message)

    await app(scope, receive, send)
    return messages


if __name__ == "__main__":
    # Demostración del contrato WSGI
    status, body = inspect_wsgi_contract(wsgi_app.wsgi_app)
    print(f"WSGI status: {status}")

    # Demostración del contrato ASGI
    messages = asyncio.run(inspect_asgi_contract(asgi_app))
    response_start = next(m for m in messages if m["type"] == "http.response.start")
    print(f"ASGI status: {response_start['status']}")

Qué está pasando realmente aquí

El punto de entrada de inspect_wsgi_contract te muestra el contrato en su forma más desnuda: la aplicación es una función que el servidor llama una vez, de forma síncrona, y cuando retorna la respuesta está completa. No hay forma de que la app diga “espera, todavía tengo más datos”; la única salida es el iterable que devuelve.

inspect_asgi_contract revela por qué ASGI puede hacer más: la app recibe send como herramienta para empujar mensajes al servidor en cualquier momento. Fíjate en cómo event_stream usa yield en un generador async: cada yield llama a send internamente vía StreamingResponse, manteniendo la conexión abierta y enviando datos cuando están listos. Eso no es un truco de FastAPI, es lo que el protocolo ASGI hace posible por diseño.

El endpoint /asgi/fan-out ilustra el caso más común donde ASGI gana de forma objetiva. Con WSGI y Flask necesitarías concurrent.futures.ThreadPoolExecutor para hacer tres llamadas HTTP en paralelo, lo cual funciona pero añade complejidad y consume threads del OS. Con ASGI el fan-out es asyncio.gather: un solo thread del OS, sin context switches de kernel, con el mismo efecto.

La elección de servidor refleja directamente esta dicotomía. Gunicorn con --workers 4 arranca cuatro procesos Python; cada uno puede procesar una request bloqueante simultáneamente. Uvicorn con --workers 1 arranca un proceso con un event loop que maneja miles de conexiones concurrentes siempre que el I/O sea async. En producción se suele combinar Gunicorn como gestor de procesos con Uvicorn como worker class (gunicorn -k uvicorn.workers.UvicornWorker), obteniendo fault isolation de procesos más concurrencia dentro de cada uno.

El overhead de ASGI que mencioné antes es concreto: el event loop de asyncio tiene latencia de scheduling, el modelo mental de “todo debe ser async o bloqueas el loop” es una restricción real, y librerías síncronas como psycopg2 o el ORM clásico de Django bloquean el worker si las llamas sin asyncio.run_in_executor. Si tu app hace queries SQL, sirve HTML renderizado y no tiene WebSockets ni I/O masivamente paralelo, Flask + Gunicorn te evita esa clase de accidentes.

Errores que debes conocer

Error: llamar código síncrono bloqueante desde un handler async, matando la concurrencia que ASGI prometía.

# ❌ Wrong
import psycopg2

@asgi_app.get("/users")
async def get_users():
    conn = psycopg2.connect("dbname=mydb")   # bloquea el event loop entero
    cur = conn.cursor()
    cur.execute("SELECT * FROM users")
    return cur.fetchall()

# ✅ Right
import asyncpg

@asgi_app.get("/users")
async def get_users():
    conn = await asyncpg.connect("postgresql://localhost/mydb")
    rows = await conn.fetch("SELECT * FROM users")
    await conn.close()
    return [dict(r) for r in rows]

Usar un driver async (asyncpg, aiosqlite, aiomysql) garantiza que la llamada cede al event loop mientras espera la base de datos.

Error: desplegar una app ASGI con un servidor WSGI puro (o viceversa), lo que produce errores crípticos en runtime en lugar de un fallo claro al arrancar.

# ❌ Wrong: gunicorn no entiende el contrato ASGI
gunicorn myapp:asgi_app

# ✅ Right: uvicorn para ASGI, o gunicorn con el worker class correcto
uvicorn myapp:asgi_app --host 0.0.0.0 --port 8000
# o en producción multi-worker:
gunicorn myapp:asgi_app -k uvicorn.workers.UvicornWorker -w 4

El worker class de Uvicorn dentro de Gunicorn es el puente: Gunicorn gestiona los procesos y las señales del OS, Uvicorn implementa el protocolo ASGI dentro de cada proceso.

N° 180