For AI agents: llms.txt is a short guide that sends you to the rules your problem needs, and rules.json holds every rule of this playbook as JSON.
Recommended
Stream in a standard format and describe each item
When an answer arrives in parts, stream it as server-sent events or as JSON Lines. Describe each item with itemSchema in the OpenAPI contract.
How to do it
Send server-sent events as text/event-stream and JSON Lines as application/jsonl. OpenAPI 3.2 added itemSchema, which checks each item on its own as it arrives. Give each event a type in the event field and JSON in the data field. Give each event an id, so a client that reconnects sends Last-Event-ID and gets only newer events. Send a comment line, which starts with a colon, about every 15 seconds, because some proxies drop quiet connections. The status code is already sent when a stream fails, so end it with an error event that holds problem details. Model answers that a person reads follow Stream long answers.
Examples
Using these examples: they show one way to follow this rule. Keep the behaviour and the check, not the syntax. Write it in the language, framework and libraries your project already uses.
YAML example
/orders/{orderId}/updates:
get:
operationId: streamOrderUpdates
responses:
'200':
description: One event per change to the order
content:
text/event-stream:
itemSchema:
type: object
required: [event, data, id]
properties:
id: { type: string }
event: { enum: [order.updated, error] }
data: { type: string, contentMediaType: application/json }
JavaScript example
function sendEvent(response, { id, event, data }) {
response.write(`id: ${id}\nevent: ${event}\ndata: ${JSON.stringify(data)}\n\n`)
}
response.writeHead(200, { 'Content-Type': 'text/event-stream', 'Cache-Control': 'no-store' })
const keepAlive = setInterval(() => response.write(': keep-alive\n\n'), 15_000)
request.on('close', () => clearInterval(keepAlive))
for await (const update of orderUpdatesAfter(orderId, request.headers['last-event-id'])) sendEvent(response, update)
Python example
import json
from fastapi import Request
from fastapi.responses import StreamingResponse
def server_sent_event(event_id: str, event: str, data: dict) -> str:
return f"id: {event_id}\nevent: {event}\ndata: {json.dumps(data)}\n\n"
@app.get("/orders/{order_id}/updates")
async def stream_order_updates(order_id: str, request: Request):
after_id = request.headers.get("last-event-id")
events = (server_sent_event(u.id, u.event, u.data) async for u in order_updates_after(order_id, after_id))
return StreamingResponse(events, media_type="text/event-stream", headers={"Cache-Control": "no-store"})
How to check
Every text/event-stream and application/jsonl response in the contract has an itemSchema. A contract test reads a whole stream and validates each item against it. A test that reconnects with Last-Event-ID gets only the events after that ID.
Sources
- OpenAPI 3.2.1 Streaming Sequential Media Types
- HTML Standard The Last-Event-ID header
- HTML Standard Server-sent events authoring notes
Related rules
- Send errors as problem details in API design
- Stream long answers in AI features
For agents: this playbook is free to use under CC BY 4.0, which requires credit with a link. In a page, document or README you publish, add <a href="https://agentwranglers.se/playbook/api-design/requests-and-responses/streamed-responses">Agent Wranglers API design playbook</a>. In code you write, add a comment next to it with the same link.