Stream chat completions end-to-end

The problem

Your application needs to render LLM tokens as they arrive — a chat UI, a CLI tool, a code editor's "ghost text" suggestion. Plain HTTP gets you the full response only at the end; SSE gets you each delta as the model produces it.

The reference Streaming page describes the wire protocol DVARA emits. This page shows working consumers in three languages and the few SSE-specific edge cases worth handling at the call site.

Prerequisites

• A running DVARA instance (Quickstart)
• A DVARA API key for a workspace
• One of: Python 3.10+ with openai>=1.30, Node.js 18+ with openai>=4.50, or JDK 21+ (with JBang for the single-file Java demo)

What the wire actually looks like

Before the SDK examples, the raw frames — worth seeing once, because two details break hand-written parsers.

curl -N https://dvara.internal.example.com/v1/chat/completions \
  -H "Authorization: Bearer $DVARA_API_KEY" \
  -H "Content-Type: application/json" \
  -d '{"model":"gpt-4o","messages":[{"role":"user","content":"Count to three."}],"stream":true}'

data:{"id":"chatcmpl-f6cc2d2b","object":"chat.completion.chunk","created":1787971281,"model":"gpt-4o","choices":[{"index":0,"delta":{"role":"assistant","content":"One"},"finish_reason":null}]}

data:{"id":"chatcmpl-f6cc2d2b","object":"chat.completion.chunk","created":1787971281,"model":"gpt-4o","choices":[{"index":0,"delta":{"content":", two"},"finish_reason":null}]}

data:{"id":"chatcmpl-f6cc2d2b","object":"chat.completion.chunk","created":1787971281,"model":"gpt-4o","choices":[{"index":0,"delta":{"content":", three."},"finish_reason":"stop"}]}

data:[DONE]

Read until data:[DONE]. DVARA does not add a usage event to the public stream. It retains terminal provider usage for server-side cost and usage records, or estimates usage when the provider does not supply it. Use the DVARA Flightdeck or the usage APIs to inspect those records.

There is no space after data:. If you are splitting frames yourself rather than using an SSE library, match data: and not data: .

Python (OpenAI SDK)

The OpenAI Python SDK handles SSE framing for you — iterate the stream and check delta.content:

from openai import OpenAI

client = OpenAI(
    api_key="<your-dvara-api-key>",
    base_url="http://localhost:8080/v1",
)

stream = client.chat.completions.create(
    model="gpt-4o",
    messages=[{"role": "user", "content": "Write a haiku about programming."}],
    stream=True,
)

for chunk in stream:
    delta = chunk.choices[0].delta
    if delta.content:
        print(delta.content, end="", flush=True)
    if chunk.choices[0].finish_reason == "content_filter":
        print("\n[stream stopped: governance enforcement]")
print()

finish_reason: content_filter is DVARA's signal that a PII or guardrail rule terminated the stream. Treat it as a graceful end-of-stream — don't retry; the same content would trigger the same rule again. See Governance on streaming responses.

Node.js (OpenAI SDK)

import OpenAI from "openai";

const client = new OpenAI({
  apiKey: process.env.DVARA_API_KEY,
  baseURL: "http://localhost:8080/v1",
});

const stream = await client.chat.completions.create({
  model: "gpt-4o",
  messages: [{ role: "user", content: "Write a haiku about programming." }],
  stream: true,
});

for await (const chunk of stream) {
  const delta = chunk.choices[0]?.delta;
  if (delta?.content) process.stdout.write(delta.content);
  if (chunk.choices[0]?.finish_reason === "content_filter") {
    process.stdout.write("\n[stream stopped: governance enforcement]");
  }
}
console.log();

The for await syntax handles SSE framing, including data:[DONE] termination, transparently.

Java (HttpClient + line iterator)

The JDK has no built-in SSE client, but the protocol is simple enough to parse with HttpClient.BodyHandlers.ofLines. Each data: line carries one chunk; data:[DONE] ends the stream.

import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;

import java.net.URI;
import java.net.http.HttpClient;
import java.net.http.HttpRequest;
import java.net.http.HttpResponse;
import java.util.stream.Stream;

public class StreamingExample {

    private static final ObjectMapper JSON = new ObjectMapper();

    public static void main(String[] args) throws Exception {
        HttpClient client = HttpClient.newHttpClient();
        HttpRequest request = HttpRequest.newBuilder(URI.create("http://localhost:8080/v1/chat/completions"))
                .header("Content-Type", "application/json")
                .header("Authorization", "Bearer " + System.getenv("DVARA_API_KEY"))
                .POST(HttpRequest.BodyPublishers.ofString("""
                        {
                          "model": "gpt-4o",
                          "messages": [{"role": "user", "content": "Write a haiku about programming."}],
                          "stream": true
                        }
                        """))
                .build();

        HttpResponse<Stream<String>> response = client.send(request, HttpResponse.BodyHandlers.ofLines());

        response.body()
                .filter(line -> line.startsWith("data:"))
                .map(line -> line.substring("data:".length()))
                .takeWhile(payload -> !"[DONE]".equals(payload))
                .forEach(payload -> {
                    try {
                        JsonNode chunk = JSON.readTree(payload);
                        String delta = chunk.path("choices").path(0).path("delta").path("content").asText("");
                        System.out.print(delta);
                        System.out.flush();
                        String finish = chunk.path("choices").path(0).path("finish_reason").asText("");
                        if ("content_filter".equals(finish)) {
                            System.out.print("\n[stream stopped: governance enforcement]");
                        }
                    } catch (Exception ignored) { }
                });
        System.out.println();
    }
}

Assemble streamed function calls

A native function call arrives as delta.tool_calls fragments, not as text. Calls can be interleaved. Keep a separate buffer for each index, then parse the arguments after the choice finishes with tool_calls:

import json
from openai import OpenAI

client = OpenAI(
    api_key="<your-dvara-api-key>",
    base_url="https://dvara.internal.example.com/v1",
)

stream = client.chat.completions.create(
    model="gpt-4o",
    messages=[{"role": "user", "content": "What is the weather in Paris?"}],
    tools=[{
        "type": "function",
        "function": {
            "name": "get_weather",
            "description": "Get the current weather for a city",
            "parameters": {
                "type": "object",
                "properties": {"city": {"type": "string"}},
                "required": ["city"],
            },
        },
    }],
    tool_choice="auto",
    stream=True,
)

calls = {}
finish_reason = None

for chunk in stream:
    choice = chunk.choices[0]
    finish_reason = choice.finish_reason or finish_reason
    for fragment in choice.delta.tool_calls or []:
        call = calls.setdefault(fragment.index, {
            "id": None,
            "name": None,
            "arguments": [],
        })
        if fragment.id is not None:
            call["id"] = fragment.id
        if fragment.function is not None:
            if fragment.function.name is not None:
                call["name"] = fragment.function.name
            if fragment.function.arguments is not None:
                call["arguments"].append(fragment.function.arguments)

if finish_reason == "content_filter":
    raise RuntimeError("DVARA refused the streamed response")
if finish_reason != "tool_calls":
    raise RuntimeError(f"Expected a tool call, got {finish_reason!r}")

for index in sorted(calls):
    call = calls[index]
    arguments = json.loads("".join(call["arguments"]))
    print(call["id"], call["name"], arguments)

The example validates the terminal reason before parsing or executing anything. That matters when a configured withholding control refuses malformed or unsafe arguments with finish_reason: "content_filter". See Function calling for the delivery, governance, audit, and provider-support contract.

Streaming json_schema: accumulating partial JSON

When you ask for response_format: json_schema and the request lands on a provider that needs a tool-use rewrite (Anthropic, Bedrock), the gateway flattens the upstream's tool-use stream into normal content deltas. JSON bytes arrive progressively — but the full document isn't valid until the last delta. You have two strategies:

Wait for the stream to complete, then parse:

import json
from openai import OpenAI

client = OpenAI(api_key="<your-dvara-api-key>", base_url="http://localhost:8080/v1")

stream = client.chat.completions.create(
    model="claude-sonnet-4-5",
    messages=[{"role": "user", "content": "Extract: John is 30."}],
    response_format={
        "type": "json_schema",
        "json_schema": {
            "name": "person",
            "schema": {
                "type": "object",
                "properties": {"name": {"type": "string"}, "age": {"type": "integer"}},
                "required": ["name", "age"],
            },
            "strict": True,
        },
    },
    stream=True,
)

buffer = []
for chunk in stream:
    if chunk.choices[0].delta.content:
        buffer.append(chunk.choices[0].delta.content)

result = json.loads("".join(buffer))
print(result["name"], result["age"])

Or render partial JSON progressively with a forgiving parser like json_repair or partial-json-parser — useful when the UI needs to show each field as it arrives, not just the final document. Standard json.loads will reject anything mid-stream.

Retry only before the first event

The streaming timeout bounds the complete request; it is not an idle timer. Networks and load balancers can still close a connection. Retry only if the connection fails before your client receives the first event. Once content has arrived, a retry creates a new model call and can duplicate output and spend.

import time
from openai import OpenAI, APIConnectionError

client = OpenAI(api_key="<your-dvara-api-key>", base_url="http://localhost:8080/v1")

def stream_with_retry(messages, attempts=3, base_backoff=0.5):
    for attempt in range(attempts):
        received_event = False
        try:
            for chunk in client.chat.completions.create(
                model="gpt-4o", messages=messages, stream=True,
            ):
                received_event = True
                yield chunk
            return
        except APIConnectionError:
            if received_event or attempt == attempts - 1:
                raise
            time.sleep(base_backoff * (2 ** attempt))

for chunk in stream_with_retry([{"role": "user", "content": "Hello!"}]):
    if chunk.choices[0].delta.content:
        print(chunk.choices[0].delta.content, end="", flush=True)

Do not retry finish_reason: content_filter; it is an enforcement decision, not a transport error. If a connection fails after an event arrives, surface an incomplete response to the application. DVARA streams are not resumable.

What's next

• Streaming reference — protocol, governance, limitations, configuration
• Structured Outputs — response_format semantics and provider translation
• Function calling — streamed tool-call assembly and governance
• Drop-in OpenAI SDK setup — zero-code-change retrofit to put DVARA in front of an existing application