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 type application/jsonl.
• FastAPI 0.134+ streams JSON Lines natively: async def f() -> AsyncIterable[Item]: yield item. It also documents the item schema in OpenAPI.
• Use StreamingResponse with 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(), then json.loads each 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

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 · Starlette 1.7.0 · Pydantic 2.13.5
• uvicorn 0.54.0 · httpx 0.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.dumps escapes newlines inside strings as \n, so a record can never break across lines. Just append a literal "\n".
• X-Accel-Buffering: no asks 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 id and 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 the X-Accel-Buffering: no header) and raise read timeouts. See the deployment setup in LLM Deployment with FastAPI + Docker + uv.
• Browsers: read JSON Lines with fetch() and response.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.