Stream Large Responses with FastAPI JSON Lines (2026)
Stream Large Responses with FastAPI JSON Lines (2026) shows how to send big result sets, exports, and long-running job output one record at a time instead of building a single huge JSON array in memory. JSON Lines (one JSON object per line, served as application/jsonl) lets the client start working on row 1 while the server is still fetching row 10,000. In 2026 FastAPI supports this natively: annotate a path operation as AsyncIterable[Model], yield items, and FastAPI validates, serializes, and streams each line for you.
PyInns already covers pushing LLM tokens to browsers with FastAPI Server-Sent Events. SSE is the right protocol for a browser EventSource that needs event names and reconnection. This guide covers the other common case: machine-to-machine streaming of structured records, like data exports, agent step logs, batch scoring results, or paginated DB reads, where every line is a complete, parseable JSON document.
TL;DR
- JSON Lines = one JSON value per line, separated by
\n. Content typeapplication/jsonl. - FastAPI 0.134+ streams JSON Lines natively:
async def f() -> AsyncIterable[Item]: yield item. It also documents the item schema in OpenAPI. - Use
StreamingResponsewith your own async generator when you need custom envelopes, an explicit end marker, or in-band error lines. - Consume incrementally with
httpx.AsyncClient.stream()+aiter_lines(), thenjson.loadseach line. - After the first byte is sent, the status code is locked at 200. Report failures inside the stream, and always clean up in
finally.
JSON Lines vs SSE vs one big JSON array
| Concern | One JSON array | JSON Lines (application/jsonl) | SSE (text/event-stream) |
|---|---|---|---|
| First byte to client | After the last item is ready | As soon as the first item is ready | As soon as the first event is ready |
| Server memory | Whole list + serialized body | One item at a time | One event at a time |
| Client parsing | response.json() | Split on newline, json.loads per line | Parse event:/data:/id: fields |
| Browser-native API | fetch().json() | fetch() + a ReadableStream reader | EventSource with auto-reconnect |
| Resume after drop | Retry everything | DIY (e.g. ?after_id=) | Built in via Last-Event-ID |
| Best fit | Small, bounded responses | Exports, ETL, logs, batch results, service-to-service | Browser UIs, chat tokens, notifications |
Rule of thumb: if the payload is small and bounded, keep returning a list. If a Python or CLI client consumes a large or open-ended sequence of records, use JSON Lines. If a browser tab listens for live events, use SSE. For two-way traffic, use WebSockets.
Versions tested (2026-09-25)
- Python
3.13.5 - FastAPI
0.141.1· Starlette1.7.0· Pydantic2.13.5 - uvicorn
0.54.0· httpx0.28.1
Native JSON Lines streaming was added in FastAPI 0.134.0 (see the official Stream JSON Lines docs). On older versions, use the StreamingResponse pattern from section 3. It works on any FastAPI release.
python3 -m venv .venv
source .venv/bin/activate
pip install "fastapi>=0.134" uvicorn httpx
uvicorn app:app --port 8000
# other terminal
python client.py http://127.0.0.1:8000
1. A model and a row source that never loads everything
The generator below stands in for a database cursor or a paginated upstream API. It yields one Order at a time. The finally block is where you would close the cursor or file handle, and it runs whether the stream finishes, fails, or the client disconnects.
import asyncio
import json
import logging
from collections.abc import AsyncIterable
from datetime import datetime, timezone
from fastapi import FastAPI, Request
from fastapi.responses import StreamingResponse
from pydantic import BaseModel
log = logging.getLogger("jsonl-demo")
logging.basicConfig(level=logging.INFO, format="%(levelname)s %(message)s")
app = FastAPI(title="PyInns JSON Lines demo")
class Order(BaseModel):
id: int
customer: str
total: float
async def fetch_orders(limit: int) -> AsyncIterable[Order]:
"""Pretend DB cursor: yields rows one by one instead of loading them all."""
try:
for i in range(1, limit + 1):
await asyncio.sleep(0.05) # simulate I/O per row / per page
yield Order(id=i, customer=f"customer-{i:03d}", total=round(i * 9.99, 2))
finally:
log.info("cursor closed (limit=%d)", limit) # release DB/file handles here
2. Native JSON Lines: AsyncIterable[Order] + yield
This is the whole endpoint. No response class, no manual json.dumps. FastAPI sees the AsyncIterable[Order] return annotation, validates each yielded item against Order, serializes it with Pydantic, and writes it as one line of application/jsonl.
# 1) Native JSON Lines (FastAPI >= 0.134): typed AsyncIterable + yield
@app.get("/orders/stream")
async def stream_orders(limit: int = 5) -> AsyncIterable[Order]:
async for order in fetch_orders(limit):
yield order
Real response from curl -sN -D - "http://127.0.0.1:8765/orders/stream?limit=2":
HTTP/1.1 200 OK
server: uvicorn
content-type: application/jsonl
Transfer-Encoding: chunked
{"id":1,"customer":"customer-001","total":9.99}
{"id":2,"customer":"customer-002","total":19.98}
A bonus of the typed version: OpenAPI describes the stream item. The generated schema for this route is {"application/jsonl": {"itemSchema": {"$ref": "#/components/schemas/Order"}}}, so client generators and /docs can tell what each line contains. For a blocking data source, use a plain def with Iterable[Order], and FastAPI runs it without blocking the event loop.
3. Manual StreamingResponse: envelopes, end marker, in-band errors
The native style covers most cases. Drop down to StreamingResponse when the protocol needs more than bare items: a type field so the client can tell rows from control messages, an explicit end record (so a truncated stream can’t be mistaken for a complete one), and an error record when something breaks halfway.
# 2) Manual StreamingResponse: full control over lines, errors, disconnects
async def order_lines(request: Request, limit: int, fail_at: int | None):
sent = 0
try:
async for order in fetch_orders(limit):
if await request.is_disconnected():
log.info("client disconnected after %d rows - stopping query", sent)
break
if fail_at is not None and order.id == fail_at:
raise RuntimeError(f"upstream failed on row {order.id}")
yield json.dumps({"type": "row", "data": order.model_dump()}) + "\n"
sent += 1
else:
yield json.dumps({"type": "end", "rows": sent}) + "\n"
except asyncio.CancelledError:
log.info("stream cancelled by server after %d rows (client went away)", sent)
raise # never swallow cancellation
except Exception as exc: # headers already sent: report in-band, then stop
log.warning("stream error after %d rows: %s", sent, exc)
yield json.dumps({"type": "error", "rows": sent, "detail": str(exc),
"at": datetime.now(timezone.utc).isoformat()}) + "\n"
finally:
log.info("generator closed (sent=%d)", sent)
@app.get("/orders/export")
async def export_orders(request: Request, limit: int = 5, fail_at: int | None = None):
return StreamingResponse(
order_lines(request, limit, fail_at),
media_type="application/jsonl",
headers={"Cache-Control": "no-cache", "X-Accel-Buffering": "no"},
)
Key details:
- One line per record.
json.dumpsescapes newlines inside strings as\n, so a record can never break across lines. Just append a literal"\n". X-Accel-Buffering: noasks nginx not to buffer the response, so lines reach the client as they are produced. Without it, a reverse proxy can hold back the whole “stream”.- Never swallow
CancelledError. Log it, then re-raise so the server can finish tearing the task down.
4. Consume it incrementally with httpx aiter_lines()
client.stream() returns as soon as headers arrive, and aiter_lines() yields each decoded line as soon as it is complete. The client never holds the full body. Set a read timeout: for streams it limits the gap between chunks, not the total duration.
"""Consume JSON Lines incrementally with httpx.AsyncClient + aiter_lines()."""
import asyncio
import json
import sys
import time
import httpx
BASE = sys.argv[1] if len(sys.argv) > 1 else "http://127.0.0.1:8000"
async def consume(path: str, stop_after: int | None = None) -> None:
t0 = time.perf_counter()
timeout = httpx.Timeout(10.0, read=30.0) # read timeout = max gap between chunks
async with httpx.AsyncClient(base_url=BASE, timeout=timeout) as client:
async with client.stream("GET", path) as resp:
resp.raise_for_status()
print(f"GET {path} -> {resp.status_code} {resp.headers['content-type']}")
count = 0
async for line in resp.aiter_lines():
if not line.strip():
continue
record = json.loads(line)
count += 1
print(f" +{time.perf_counter() - t0:5.2f}s {record}")
if record.get("type") == "error":
print(" !! server reported an error mid-stream; partial data kept")
break
if stop_after and count >= stop_after:
print(f" -> client stops early after {count} lines")
break
async def main() -> None:
await consume("/orders/stream?limit=3")
await consume("/orders/export?limit=3")
await consume("/orders/export?limit=5&fail_at=3")
await consume("/orders/export?limit=200", stop_after=2)
await asyncio.sleep(0.5) # give the server time to log the disconnect
asyncio.run(main())
Real output from the tested run. Note the timestamps: rows arrive about 50 ms apart as the server produces them, not all at once at the end.
GET /orders/stream?limit=3 -> 200 application/jsonl
+ 0.10s {'id': 1, 'customer': 'customer-001', 'total': 9.99}
+ 0.16s {'id': 2, 'customer': 'customer-002', 'total': 19.98}
+ 0.21s {'id': 3, 'customer': 'customer-003', 'total': 29.97}
GET /orders/export?limit=3 -> 200 application/jsonl
+ 0.06s {'type': 'row', 'data': {'id': 1, 'customer': 'customer-001', 'total': 9.99}}
+ 0.11s {'type': 'row', 'data': {'id': 2, 'customer': 'customer-002', 'total': 19.98}}
+ 0.16s {'type': 'row', 'data': {'id': 3, 'customer': 'customer-003', 'total': 29.97}}
+ 0.16s {'type': 'end', 'rows': 3}
GET /orders/export?limit=5&fail_at=3 -> 200 application/jsonl
+ 0.06s {'type': 'row', 'data': {'id': 1, 'customer': 'customer-001', 'total': 9.99}}
+ 0.11s {'type': 'row', 'data': {'id': 2, 'customer': 'customer-002', 'total': 19.98}}
+ 0.16s {'type': 'error', 'rows': 2, 'detail': 'upstream failed on row 3', 'at': '2026-09-25T12:59:08.064681+00:00'}
!! server reported an error mid-stream; partial data kept
GET /orders/export?limit=200 -> 200 application/jsonl
+ 0.06s {'type': 'row', 'data': {'id': 1, 'customer': 'customer-001', 'total': 9.99}}
+ 0.11s {'type': 'row', 'data': {'id': 2, 'customer': 'customer-002', 'total': 19.98}}
-> client stops early after 2 lines
5. Error handling mid-stream: what really happens
Once the first line is sent, the HTTP status (200) and headers are already on the wire. You can’t switch to a 500 later. That leaves two outcomes, and the demo shows both:
- Manual generator with
try/except: the client receives the good rows plus{"type": "error", "rows": 2, "detail": "upstream failed on row 3", ...}, as in the output above. The client knows exactly how much data is valid. - Native endpoint that raises: the server aborts the chunked response. The client gets the lines that were already sent, then a transport error. Tested with the handler below:
@app.get("/boom")
async def boom() -> AsyncIterable[Row]:
for i in range(1, 5):
await asyncio.sleep(0.05)
if i == 3:
raise RuntimeError("db died")
yield Row(id=i)
line {"id":1}
line {"id":2}
RemoteProtocolError peer closed connection without sending complete message body (incomplete chunked read)
Both behaviours are safe in their own way. An unexpected crash never looks like a clean end, because httpx raises RemoteProtocolError. For exports that people rely on, prefer explicit end/error records so clients can tell “finished” from “cut off” without depending on transport errors. Also validate query parameters and permissions before the first yield, while you can still return a normal 4xx.
6. Disconnects and backpressure
Streaming is pull-based. await send() only completes when the ASGI server has handed the bytes to the socket, so a slow client naturally slows down the generator. That is your backpressure, as long as you don’t pre-load data into an unbounded queue. When the client leaves early, you want the database query to stop too. Server log from the run where the client broke out after 2 lines:
INFO: 127.0.0.1:51414 - "GET /orders/export?limit=200 HTTP/1.1" 200 OK
INFO cursor closed (limit=200)
INFO stream cancelled by server after 2 rows (client went away)
INFO generator closed (sent=2)
INFO: 127.0.0.1:51424 - "GET /orders/stream?limit=200 HTTP/1.1" 200 OK
INFO cursor closed (limit=200)
On the tested stack (Starlette 1.7 + uvicorn 0.54), the framework noticed the disconnect and cancelled the generator during its next await. The CancelledError branch logged it, and both finally blocks closed the “cursor”. The native endpoint cleaned up the same way. The await request.is_disconnected() check in the loop is still worth its one line. It is a cheap guard before each expensive step (next DB page, next LLM call) and protects you on servers or middleware stacks that don’t cancel promptly. Just don’t rely on it as your only cleanup path. finally is what always runs.
7. The anti-pattern, for comparison
# 3) The anti-pattern for large data: build one giant JSON array in memory
@app.get("/orders/all")
async def all_orders(limit: int = 5) -> list[Order]:
return [o async for o in fetch_orders(limit)]
This returns application/json only after every row is fetched and the whole list is serialized. With 3 rows that’s fine. With 3 million, memory usage and time to first byte grow with the result size, and proxies may time out before you send anything.
8. Smoke tests with TestClient
demo/test_app.py checks content types, the end and error records, the OpenAPI item schema, and the array endpoint. Output:
native : application/jsonl {'id': 1, 'customer': 'customer-001', 'total': 9.99}
manual : {'type': 'end', 'rows': 3}
error : {'type': 'error', 'rows': 2, 'detail': 'upstream failed on row 3'}
openapi: {"application/jsonl": {"itemSchema": {"$ref": "#/components/schemas/Order"}}}
array : application/json 3 items (sent only when all are ready)
ALL PASSED
Heads-up: with Starlette 1.7 the test client logs a StarletteDeprecationWarning recommending the httpx2 package for TestClient. The tests still pass with httpx 0.28.1. Pin whichever your CI uses.
Production tips
- Keep lines small and flat. Stream one row, one event, or one result per line, not nested pages of 10,000 rows.
- Add resume support yourself. Include a monotonically increasing
idand accept?after_id=, so a client can reconnect and continue from its last good line. - Heartbeats for slow producers. If a line can take longer than proxy idle timeouts (often 60 s), yield a small
{"type": "ping"}record periodically. - Compression: gzip middleware buffers output, which can delay lines. Test it with your proxy, or compress only non-streaming routes.
- Behind Docker / a reverse proxy: disable proxy buffering for the streaming route (
proxy_buffering off;or theX-Accel-Buffering: noheader) and raise read timeouts. See the deployment setup in LLM Deployment with FastAPI + Docker + uv. - Browsers: read JSON Lines with
fetch()andresponse.body.getReader(), splitting on newlines. If you want HTML fragments instead of JSON, see the no-build FastAPI + HTMX approach.
Wrap-up
For large or open-ended result sets, JSON Lines is the simplest streaming format that stays easy to parse. In FastAPI 0.134+ it is one annotation away: -> AsyncIterable[Model] plus yield. Use StreamingResponse when you need end and error records, consume with httpx aiter_lines(), put cleanup in finally, and let cancellation do its job. When the consumer is a browser tab waiting for live tokens, switch to FastAPI SSE. When you ship it, follow the container setup in LLM Deployment with FastAPI + Docker + uv.