Async LLM calls in Flask with asyncio and Quart

Flask’s WSGI model ties each request to a worker thread, so a slow upstream LLM call stalls that worker until the response returns. Swapping the framework for Quart lets you write async request handlers and await non-blocking I/O, which is the clean way to issue flask asyncio quart llm calls without spinning up thread pools. This how-to builds a small service that sends multiple prompts to an OpenAI-compatible endpoint in parallel and aggregates the outputs.

Step 1: Install Quart and an async HTTP client

Quart implements the Flask API on top of asyncio, so existing Flask knowledge transfers directly. Install it alongside an async HTTP client and the official OpenAI SDK, which supports async out of the box.

python -m venv venv
source venv/bin/activate
pip install quart httpx openai

Use Python 3.9 or newer. The event loop and asyncio primitives behave predictably there, and asyncio.TaskGroup (3.11+) or gather give you structured concurrency. Avoid Python 3.7—its asyncio deprecations will bite mid-project.

Step 2: Write a minimal Quart application

Create app.py with a single health route. Quart’s Quart class mirrors Flask’s constructor, and @app.route works identically except the view function is async def.

from quart import Quart, jsonify

app = Quart(__name__)

@app.route("/health")
async def health():
    return jsonify(status="ok")

if __name__ == "__main__":
    app.run()

Run with quart run or an ASGI server like Hypercorn. The server now handles requests on the event loop; an await inside a view yields control instead of blocking the thread. That yield is what makes flask asyncio quart llm calls scalable on a single process.

Step 3: Issue a single async LLM call

The OpenAI Python client exposes AsyncOpenAI. Point it at any OpenAI-compatible endpoint. The key is awaiting client.chat.completions.create instead of calling it synchronously.

from openai import AsyncOpenAI

client = AsyncOpenAI(
    base_url="https://api.openai.com/v1",
    api_key="sk-...",
)

@app.route("/complete/<prompt>")
async def complete(prompt: str):
    resp = await client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{"role": "user", "content": prompt}],
        max_tokens=128,
    )
    return jsonify(text=resp.choices[0].message.content)

This view returns immediately to the event loop while the HTTP request is in flight. Under load, one worker can serve many pending LLM calls because it never sleeps on a socket.

Step 4: Fire parallel flask asyncio quart llm calls

Real workloads rarely need one completion; they need ten. Use asyncio.gather to dispatch concurrent requests. The event loop multiplexes them over a shared connection pool.

import asyncio
from quart import request

@app.route("/batch", methods=["POST"])
async def batch():
    data = await request.get_json()
    prompts = data.get("prompts", [])

    async def call_one(p):
        r = await client.chat.completions.create(
            model="gpt-4o-mini",
            messages=[{"role": "user", "content": p}],
            max_tokens=128,
        )
        return r.choices[0].message.content

    results = await asyncio.gather(*(call_one(p) for p in prompts))
    return jsonify(results=results)

If you pass ["explain asyncio", "what is Quart?"], both requests leave simultaneously. Total latency approximates the slowest call, not the sum. That difference is the entire point of flask asyncio quart llm calls.

Why not threads?

Flask with threaded=True can parallelize blocking calls, but each thread holds a stack and GIL contention reduces throughput. Async IO collapses that to cooperative scheduling; you write linear code with await and the loop does the multiplexing. For I/O-bound LLM calls, this is strictly more efficient and easier to reason about.

Step 5: Add timeouts and resilient fallback

Network calls fail. Wrap each request with asyncio.wait_for to bound latency. If you point the client at a gateway such as n4n.ai’s OpenAI-compatible endpoint, you also get automatic fallback when a provider is rate-limited or degraded, without extra code. Otherwise, implement a simple retry.

async def call_one(p, timeout=10.0):
    try:
        return await asyncio.wait_for(call_llm(p), timeout)
    except asyncio.TimeoutError:
        return f"timeout on: {p}"

async def call_llm(p):
    return (await client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{"role": "user", "content": p}],
        max_tokens=128,
    )).choices[0].message.content

Set HTTPX timeouts on the client too:

from openai import AsyncOpenAI
import httpx

client = AsyncOpenAI(
    base_url="https://api.openai.com/v1",
    api_key="sk-...",
    http_client=httpx.AsyncClient(timeout=httpx.Timeout(10.0)),
)

Never let an unbounded await sit in a request handler. A stuck upstream will silently exhaust your event loop’s pending tasks.

Step 6: Stream tokens to the client

For chat UIs, streaming beats waiting. Quart supports async generators as response bodies. Use the SDK’s stream=True and yield chunks.

from quart import Response, stream_with_context

@app.route("/stream/<prompt>")
async def stream(prompt: str):
    @stream_with_context
    async def generate():
        stream = await client.chat.completions.create(
            model="gpt-4o-mini",
            messages=[{"role": "user", "content": prompt}],
            max_tokens=256,
            stream=True,
        )
        async for chunk in stream:
            if chunk.choices[0].delta.content:
                yield chunk.choices[0].delta.content
    return Response(generate(), mimetype="text/plain")

The browser receives tokens as they arrive. The event loop keeps serving other requests between chunks, so one slow generation doesn’t block the whole service.

Step 7: Run and verify success

Quart needs an ASGI server. Hypercorn is the reference choice.

pip install hypercorn
hypercorn app:app --bind 0.0.0.0:8000

Verify with curl. First, health:

curl http://localhost:8000/health
# {"status":"ok"}

Then a single completion:

curl http://localhost:8000/complete/hello
# {"text":"..."}

Then the batch endpoint with parallel flask asyncio quart llm calls:

curl -X POST http://localhost:8000/batch \
  -H "Content-Type: application/json" \
  -d '{"prompts":["write a haiku","summarize asyncio"]}'
# {"results":["...","..."]}

Measure timing with time curl on the batch versus sequential calls; the batch should be near the max of individual latencies, proving concurrency. For streaming:

curl http://localhost:8000/stream/why%20async
# tokens print incrementally

If the batch endpoint returns in roughly the same time as a single call, your flask asyncio quart llm calls are truly parallel.

Step 8: Structure for a larger service

When your flask asyncio quart llm calls grow beyond a prototype, split routes into Blueprints and inject the client via app config.

from quart import Blueprint, current_app

llm_bp = Blueprint("llm", __name__)

@llm_bp.route("/batch", methods=["POST"])
async def batch():
    client = current_app.config["OPENAI_CLIENT"]
    # ... same gather logic

Register it in app.py:

from app.routes import llm_bp
app.register_blueprint(llm_bp)
app.config["OPENAI_CLIENT"] = AsyncOpenAI(api_key="sk-...")

Keep the client at module or app level. Creating one per request leaks connections and defeats the pooled transport. For tests, use pytest-asyncio and httpx_mock to stub the endpoint so you never hit a real model in CI.

Operational notes

• Set AsyncOpenAI(max_retries=0) if you handle retries yourself; the SDK retries by default.
• If you deploy behind a reverse proxy, ensure it supports ASGI (Nginx Unit, Traefik with Hypercorn, or Uvicorn workers).
• Honor provider cache-control hints by passing extra_headers if your gateway forwards them; it cuts cost on repeated prefixes.

Following these steps gives you a production-shaped Quart service that makes non-blocking flask asyncio quart llm calls, scales on a single process, and degrades gracefully when upstream models misbehave.