Dropped connections are inevitable when you stream LLM completions over HTTP. Implementing sse reconnect last-event-id correctly lets your client resume a truncated stream without re-requesting the entire prompt, saving tokens and latency.
Step 1: Understand the SSE resume contract
The SSE spec defines a simple mechanism: each event may carry an id: field. The client remembers the last id received. If the connection closes, the client waits for the retry: period (default 3 seconds) and re-issues the GET with a Last-Event-ID HTTP header set to that id.
A server that respects this header can truncate or replay from that point. For LLM token streams, you typically assign id: to each chunk—either the token index or a UUID per chunk. The sse reconnect last-event-id flow is not magic: the server must do something sensible with the header, or the client must dedupe.
Step 2: Emit event IDs from your streaming endpoint
Below is a minimal FastAPI route that streams completion-like text with per-chunk IDs. We use a generator that yields SSE-formatted bytes.
from fastapi import FastAPI, Request
from fastapi.responses import StreamingResponse
import asyncio, json
app = FastAPI()
async def fake_llm_tokens(prompt: str):
for i, tok in enumerate(["Hello", " ", "world", "!", ""]):
await asyncio.sleep(0.1)
yield f"id: {i}\ndata: {json.dumps({'token': tok})}\n\n"
@app.get("/stream")
async def stream(request: Request):
return StreamingResponse(
fake_llm_tokens(request.query_params.get("prompt", "")),
media_type="text/event-stream",
headers={"Cache-Control": "no-cache", "X-Accel-Buffering": "no"}
)The id: {i} line is what the client will echo back as Last-Event-ID.
Step 3: Capture Last-Event-ID in a non-browser client
Browser EventSource handles this automatically. Server-side or CLI clients usually do not. Here is a Python aiohttp client that parses the stream, stores the last id, and reconnects with the header.
import aiohttp, asyncio
async def stream_with_reconnect(url, max_retries=5):
last_id = None
attempt = 0
while attempt < max_retries:
headers = {}
if last_id is not None:
headers["Last-Event-ID"] = str(last_id)
try:
async with aiohttp.ClientSession() as session:
async with session.get(url, headers=headers) as resp:
async for line in resp.content:
line = line.decode().strip()
if line.startswith("id:"):
last_id = int(line[3:].strip())
elif line.startswith("data:"):
print("TOKEN:", line[5:].strip())
except (aiohttp.ClientError, asyncio.TimeoutError) as e:
print(f"drop detected: {e}, reconnecting from {last_id}")
attempt += 1
await asyncio.sleep(min(2 ** attempt, 30))
continue
breakThis loop implements sse reconnect last-event-id by persisting last_id across connection failures.
Step 4: Server-side handling of Last-Event-ID
Most LLM providers cannot resume a generation mid-sequence. If you operate the endpoint, you have two options:
- Dedupe on client: Ignore the header server-side, but ensure IDs are monotonic. The client drops any event with
id <= last_idafter reconnect. - Context rebuild: Use the header to know how many tokens the client already has, then re-request the completion with prior text appended.
A gateway such as n4n.ai forwards provider cache-control hints and honors routing directives, but it will not buffer an entire stream for you; your client owns resume logic.
Minimal server that skips already-sent chunks based on the header:
@app.get("/stream")
async def stream(request: Request):
last_id = request.headers.get("Last-Event-ID")
start = int(last_id) + 1 if last_id is not None else 0
async def gen():
for i, tok in enumerate(["Hello", " ", "world", "!", ""]):
if i < start:
continue
await asyncio.sleep(0.1)
yield f"id: {i}\ndata: {json.dumps({'token': tok})}\n\n"
return StreamingResponse(gen(), media_type="text/event-stream")Step 5: Set retry hints and backoff
The SSE spec lets the server send retry: <ms> to control client reconnect delay. If you use a custom client, implement exponential backoff with jitter.
import random
def backoff(attempt):
return min(2 ** attempt, 30) + random.uniform(0, 1)Send a retry directive from the server occasionally:
yield f"retry: 2000\n\n"Step 6: Avoid duplicate side effects
If your stream triggers downstream actions (e.g., logging per token), reconnection can cause duplicates. Use the event ID as an idempotency key. Store processed IDs in a set or Redis.
processed = set()
if event_id not in processed:
processed.add(event_id)
handle_token(data)Step 7: Verify the reconnection end to end
You need a way to forcibly drop the TCP connection mid-stream. Run the server locally and kill it with SIGSTOP between chunks.
Test script:
uvicorn main:app --port 8000 &
SERVER_PID=$!
sleep 2
python client.py &
CLIENT_PID=$!
sleep 0.5
kill -STOP $SERVER_PID
sleep 2
kill -CONT $SERVER_PID
wait $CLIENT_PID
kill $SERVER_PIDSuccess criteria: the client prints each token exactly once, and the Last-Event-ID header appears in the server logs on the second request. If you see repeated tokens, your dedupe or skip logic is broken.
Step 8: Production considerations
- ID format: Use integer counters for simplicity; UUIDs if you shard streams.
- Header size:
Last-Event-IDis a single string; keep it small. - TLS: SSE over HTTPS works; ensure your client respects cert errors.
- Gateways: When calling an OpenAI-compatible endpoint, note that the upstream may not emit
id:fields. You can wrap it: assign IDs client-side by counting chunks, then apply sse reconnect last-event-id logic against your own proxy.
The sse reconnect last-event-id pattern is low-complexity insurance against flaky networks. Implement it once in your streaming client and stop losing generations to transient 503s.