n4nAI
Open console →
API integrationTutorialPython Async/Await Streaming (asyncio)

Combining asyncio and Server-Sent Events for LLM streaming

Step-by-step tutorial on python asyncio sse llm streaming: build a client and server that stream LLM tokens over Server-Sent Events with asyncio.

n4n Team3 min read731 words

Audio narration

Coming soon — every post will get a voice note here.

Wiring up python asyncio sse llm streaming lets you push model output to browsers or internal services the moment tokens arrive, without blocking the event loop. This tutorial builds a minimal but production-shaped pipeline: an asyncio SSE server that proxies a streaming LLM endpoint, and a client that consumes the event stream.

Prerequisites

  • Python 3.10 or newer (uses async for and httpx.AsyncClient.stream).
  • aiohttp and httpx installed: pip install aiohttp httpx.
  • An OpenAI-compatible /v1/chat/completions endpoint with stream: true support. If you need one endpoint that fronts many models, n4n.ai provides an OpenAI-compatible gateway covering 240+ models with automatic fallback when a provider is degraded.

You should be comfortable with async/await and basic HTTP. No frontend framework required; we’ll test with curl and a small Python client.

SSE fundamentals

Server-Sent Events are dead simple: a long-lived HTTP response with Content-Type: text/event-stream, where each message is prefixed with data: and terminated by a blank line. No JSON envelope required, though most LLM gateways emit JSON inside the data field.

data: {"token": "Hello"}

data: {"token": " world"}

The browser EventSource API consumes this natively. For backend-to-backend hops, you parse lines yourself. SSE is unidirectional, which is exactly what you want for token streaming: the client sends one prompt, the server pushes tokens until done.

Step 1: Async LLM streaming client

We use httpx.AsyncClient.stream to avoid loading the whole response into memory. The LLM endpoint returns newline-delimited data: {json} frames, ending with data: [DONE].

import asyncio
import json
import httpx

BASE_URL = "https://api.n4n.ai"  # or your own gateway
API_KEY = "sk-..."  # read from env in real code

async def stream_llm_tokens(prompt: str, model: str = "gpt-3.5-turbo"):
    async with httpx.AsyncClient(timeout=httpx.Timeout(60.0)) as client:
        async with client.stream(
            "POST",
            f"{BASE_URL}/v1/chat/completions",
            headers={
                "Authorization": f"Bearer {API_KEY}",
                "Content-Type": "application/json",
            },
            json={
                "model": model,
                "messages": [{"role": "user", "content": prompt}],
                "stream": True,
            },
        ) as resp:
            resp.raise_for_status()
            async for line in resp.aiter_lines():
                if not line.startswith("data:"):
                    continue
                payload = line[5:].strip()
                if payload == "[DONE]":
                    break
                chunk = json.loads(payload)
                delta = chunk["choices"][0]["delta"]
                token = delta.get("content")
                if token:
                    yield token

Key detail: line[5:] strips the data: prefix (5 chars including colon). Some servers send data: with a space; .strip() handles it. If the upstream sends a comment line (: keepalive), it doesn’t start with data: and is skipped.

Parsing without blowing up

In production, wrap json.loads in a try/except. A truncated frame on connection drop should log and break, not crash the handler. For this tutorial we let errors propagate to keep the code minimal.

Step 2: Aiohttp SSE server

Aiohttp’s StreamResponse is the right primitive. Prepare it with SSE headers, then write frames as they arrive from the generator.

from aiohttp import web

async def sse_proxy(request: web.Request) -> web.StreamResponse:
    prompt = request.query.get("prompt", "Say hello in two words.")
    model = request.query.get("model", "gpt-3.5-turbo")

    resp = web.StreamResponse()
    resp.headers["Content-Type"] = "text/event-stream"
    resp.headers["Cache-Control"] = "no-cache"
    resp.headers["X-Accel-Buffering"] = "no"  # disable proxy buffering
    await resp.prepare(request)

    try:
        async for token in stream_llm_tokens(prompt, model):
            frame = f"data: {json.dumps({'token': token})}\n\n"
            await resp.write(frame.encode("utf-8"))
    except asyncio.CancelledError:
        raise  # client disconnected; let aiohttp clean up
    finally:
        await resp.write_eof()
    return resp

def create_app() -> web.Application:
    app = web.Application()
    app.router.add_get("/stream", sse_proxy)
    return app

if __name__ == "__main__":
    web.run_app(create_app(), port=8080)

The X-Accel-Buffering: no header stops nginx from buffering the response, which would defeat streaming. Without it, you’ll see tokens arrive in one giant chunk after the request ends.

Step 3: End-to-end wiring

Run the server. In another shell, consume the stream with curl:

curl -N "http://localhost:8080/stream?prompt=Explain%20asyncio%20in%20one%20sentence"

Expected output checkpoint

You should see a series of SSE frames, one per token (or per chunk), like:

data: {"token": "Asyncio"}

data: {"token": " lets"}

data: {"token": " you"}

data: {"token": " run"}

The connection stays open until the model finishes, then closes. If you point BASE_URL at a gateway that fronts multiple providers, the same code works across model swaps without changes.

Step 4: Python SSE consumer for tests

Curl is fine for a smoke test, but you’ll want a programmatic client in your test suite. Here’s a minimal async consumer using httpx:

async def consume_sse(url: str):
    async with httpx.AsyncClient(timeout=httpx.Timeout(30.0)) as client:
        async with client.stream("GET", url) as resp:
            resp.raise_for_status()
            async for line in resp.aiter_lines():
                if line.startswith("data:"):
                    payload = line[5:].strip()
                    if payload:
                        print(json.loads(payload))

Run it against your local server to assert tokens arrive incrementally. This python asyncio sse llm streaming consumer mirrors what a browser EventSource does, minus auto-reconnect logic.

Production considerations

Backpressure and cancellation

resp.write is awaitable; if the client is slow, aiohttp applies backpressure automatically. When the browser tab closes, the request task is cancelled. Catch asyncio.CancelledError to release the upstream connection—httpx.AsyncClient context manager handles that on exit. Do not suppress CancelledError silently; re-raise after cleanup.

Honoring routing and cache hints

If your upstream gateway supports routing directives, forward them. For example, n4n.ai honors client routing headers and forwards provider cache-control hints, so you can pass X-Route-To or cache-ttl headers on the client.stream call to influence provider caching. That reduces redundant token spend on repeated prefixes.

headers={
    "Authorization": f"Bearer {API_KEY}",
    "Content-Type": "application/json",
    "X-Route-To": "anthropic",  # if supported by gateway
}

Error handling on upstream

A 429 or 5xx from the model provider should map to a clean SSE comment or a custom event, not a crashed handler. Wrap the stream_llm_tokens call:

try:
    async for token in stream_llm_tokens(prompt, model):
        ...
except httpx.HTTPStatusError as e:
    await resp.write(f"event: error\ndata: {e.response.status_code}\n\n".encode())

SSE supports named events; the client can listen for addEventListener('error', ...). This keeps the stream protocol intact even when the backend fails.

Concurrency limits

Each open SSE connection holds an upstream HTTP connection. Bound your httpx connection pool (limits=httpx.Limits(max_connections=100)) and consider a semaphore around stream_llm_tokens if many clients hit the proxy at once.

Why not WebSockets?

SSE is unidirectional (server→client) and uses plain HTTP, which simplifies load balancing and works with EventSource natively. For LLM token streaming, you almost never need client→server frames after the initial prompt. If you later need bidirectional control, upgrade to WS, but you’ll lose the zero-config browser API.

Wrapping up

You now have a runnable python asyncio sse llm streaming proxy: an async httpx client that yields tokens, an aiohttp server that frames them as SSE, and both curl and Python consumers to verify behavior. Swap the BASE_URL to any OpenAI-compatible endpoint and the same code streams from 240+ models without structural changes.

Tagspythonasynciossestreaming

Written by

n4n Team

The team building n4n — a single OpenAI-compatible API in front of 240+ models, with automatic fallback, load balancing and pay-per-token metering.

More from n4n Team →
API integrationasyncio.Queue patterns for buffering streamed LLM tokensPractical patterns for using python asyncio queue streaming tokens to buffer LLM output, with code for backpressure, merging, and shutdown.API integrationDebugging asyncio deadlocks in streaming LLM applicationsA practical guide to diagnosing and fixing python asyncio deadlocks streaming llm apps, with code patterns for safe cancellation and backpressure.API integrationHow to cancel a streaming LLM request in Python asyncioLearn how to python asyncio cancel streaming request cleanly using tasks, timeouts, and async context managers to avoid token waste and socket leaks.API integrationStreaming multiple LLM models concurrently with asyncioHands-on tutorial: python asyncio stream multiple models concurrently via one OpenAI-compatible API, merging token streams safely with queues and timeouts.API integrationStreaming LLM tokens in Python with async generatorsLearn to build a production-shaped Python client for token streaming from LLMs using async generators and asyncio, with runnable code and expected output.API integrationAsync context managers for LLM API clients in PythonBuild robust Python async context managers for LLM clients that stream tokens, handle cancellation, manage connection pools, and clean up resources reliably.

All python async/await streaming (asyncio) posts →