Agent Wranglers

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

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.