Streaming LLM Responses: Server-Sent Events with Python and OpenAI-Compatible APIs

----------|----------|

First tokenWait for full generation~200ms
UXLoading spinnerText appears progressively
Timeout riskHigher (long generation)Lower
ComplexityOne requestConnection management

For chat applications and IDE tools, streaming is expected behavior.

Server Implementation (FastAPI)


from fastapi import FastAPI
from fastapi.responses import StreamingResponse
import httpx, json

app = FastAPI()

@app.get("/chat")
def chat_stream(prompt: str):
    async def generate():
        async with httpx.AsyncClient() as client:
            async with client.stream(
                "POST",
                "https://aiwave.live/v1/chat/completions",
                json={
                    "model": "deepseek-chat",
                    "messages": [{"role": "user", "content": prompt}],
                    "stream": True
                },
                headers={"Authorization": "***"}
            ) as resp:
                async for line in resp.aiter_lines():
                    if line.startswith("data: "):
                        data = json.loads(line[6:])
                        if "choices" in data:
                            delta = data["choices"][0].get("delta", {})
                            content = delta.get("content", "")
                            if content:
                                yield "data: " + json.dumps({"text": content}) + "\n\n"
                yield "data: [DONE]\n\n"
    return StreamingResponse(generate(), media_type="text/event-stream")

Client Implementation


import httpx, json

def stream_chat(prompt):
    with httpx.stream(
        "POST",
        "https://aiwave.live/v1/chat/completions",
        json={
            "model": "deepseek-chat",
            "messages": [{"role": "user", "content": prompt}],
            "stream": True
        },
        headers={"Authorization": "***"}
    ) as resp:
        for line in resp.iter_lines():
            if line.startswith("data: "):
                data = json.loads(line[6:])
                if "choices" in data:
                    content = data["choices"][0].get("delta", {}).get("content", "")
                    if content:
                        print(content, end="", flush=True)
    print()

stream_chat("Explain async/await in Python")

Key Details

  • SSE format: Each line is data: {json}\n\n. The final line is data: [DONE]\n\n
  • Stream=True: This parameter enables streaming on all OpenAI-compatible APIs
  • Works with all AIWave models: DeepSeek, GLM, Kimi, Qwen — same interface
  • Backpressure: The server sends data as fast as the model generates. If the client reads slower, the connection buffer handles it
  • Timeout: Set 30-60s for long generations
  • When to Stream

  • Chat UIs and chatbots
  • IDE autocomplete and inline suggestions
  • Any UI where users see output in real-time
  • Skip streaming for batch processing, background jobs, and API-to-API calls where latency doesn't matter.

    All models on AIWave support streaming via the standard stream: true parameter. One API key, same OpenAI-compatible format. The $1 free credit covers thousands of streaming requests to test.