Build a streaming SSE test harness across 4 LLM providers

Introduction

When you ship a streaming LLM endpoint, the bugs that surface in production are rarely the ones a JSON-shaped integration test catches. Half a frame arrives, the provider drops the connection before sending [DONE], or an Anthropic stream ends with message_stop while your parser still waits for OpenAI's terminator — and every silent failure reaches users as a hung browser tab or a truncated answer. By the end of this lesson you'll be able to design a provider-agnostic SSE test harness that validates token delivery, error propagation, and stream termination across OpenAI, Anthropic, Gemini, and Llama 4, and wires the results into a CI gate.

Key Terminology

• TTFT (Time to First Token) — elapsed time from request dispatch to the first chunk carrying non-empty content. It is the primary user-perceived latency metric your harness must measure on every streaming run, because a stream that connects but stalls is the failure mode REST tests cannot see.
• Backpressure — the condition where the consumer (your test client) reads slower than the producer (the LLM provider) sends, causing buffer accumulation that can trigger timeouts or dropped frames. Your harness needs to reproduce it deliberately so the endpoint's flow-control behavior is exercised, not just observed in production incidents.
• SSE Sentinel — the protocol-level signal indicating stream completion, such as data: [DONE] (OpenAI / Llama 4) or event: message_stop (Anthropic). Each provider terminates differently, so the harness must assert on the right sentinel per provider to catch silently-truncated streams.

Concepts

Error Injection and Edge-Case Coverage

Beyond the happy path, a robust harness must validate how your streaming endpoint behaves under failure conditions. Three categories of errors demand explicit test coverage:

• Mid-stream provider failures: The upstream LLM returns an HTTP 200 and begins streaming, then sends an SSE frame containing an error object (e.g., OpenAI's {"error": {"message": "rate limit"}}) instead of a content delta. Your endpoint must propagate this as a structured error chunk, not silently drop it.
• Connection resets: The provider closes the TCP socket before sending [DONE]. Your FastAPI endpoint must detect the incomplete stream and surface an appropriate error to the client rather than returning a truncated response that appears successful.
• Malformed SSE frames: Partial JSON payloads, missing data: prefixes, or binary garbage injected into the stream. The parse_sse_stream function's json.JSONDecodeError handler covers this, but your test must verify the handler fires and the chunk is still yielded with the _raw fallback.

To test these scenarios without hitting live providers, inject failures at the ASGI transport layer using a mock that yields crafted byte sequences. For example, a mock that sends three valid chunks followed by a connection close (no [DONE]) validates your incomplete-stream detection logic. Pair this with pytest.raises or assertion on a stream_error field in your response model.

Operational Integration with CI/CD

Streaming tests inherently depend on timing, which makes them flaky candidates for CI pipelines. Apply these hardening strategies:

For CI integration, run streaming tests against deterministic mock providers that return canned SSE sequences with fixed timing. Reserve live-provider streaming tests for a nightly or pre-release gate where flakiness is tolerable and results feed into Promptfoo evaluation dashboards for trend analysis. Tag streaming tests with @pytest.mark.streaming so teams can include or exclude them from fast feedback loops. Store TTFT percentiles (p50, p95, p99) as CI artifacts and alert when the p95 regresses beyond a configurable threshold—this catches infrastructure degradation (DNS resolution delays, TLS handshake changes, container resource limits) before it reaches users.

The streaming test harness you have built here becomes the foundation for the evaluation layers covered in the remaining goals: Promptfoo pipelines consume these same endpoints to run assertion-based prompt regression tests, RAGAS evaluation suites verify the factual quality of the streamed content, DeepEval benchmarks compare cross-provider output quality, and security red-teaming campaigns fuzz these endpoints with adversarial inputs designed to break the SSE parser or bypass content guardrails. Each subsequent layer builds on the confidence that the transport itself is reliable.

Code Walkthrough

Building on the error-injection categories and CI/CD integration patterns from the previous section, the harness comes together in three layers, each backed by code you can run against a FastAPI streaming endpoint today. First, you map the SSE contract onto each provider so the parser knows which frames mean "token", which mean "done", and which mean "error". Second, you build a StreamingTestClient that wraps httpx.AsyncClient and a parse_sse_stream() coroutine that yields uniform ParsedChunk dataclasses regardless of upstream framing. Third, you layer provider-aware assertions and fault-injection mocks on top so a single pytest invocation validates OpenAI, Anthropic, Gemini, and Llama 4/vLLM streams — including mid-stream errors and premature disconnects — without burning live tokens.

The SSE Contract and Why It Breaks Differently Per Provider

Before writing a single test, you must internalize what the SSE specification actually guarantees—and where each LLM provider deviates. The SSE protocol (defined in the W3C EventSource spec) transmits UTF-8 text frames separated by double newlines (\n\n), each containing optional event:, data:, id:, and retry: fields. The data: field carries the payload. A stream terminates when the server sends data: [DONE] (OpenAI convention) or simply closes the TCP connection.

Here is where provider divergence creates real bugs:

• OpenAI sends data: {"choices": [{"delta": {"content": "token"}}]} with a final data: [DONE] sentinel. Empty delta objects appear for role-only chunks at stream start.
• Anthropic uses a typed event system: event: content_block_delta with data: {"type": "content_block_delta", "delta": {"text": "token"}}. The stream ends with event: message_stop. There is no [DONE] sentinel.
• Gemini streams JSON objects where each chunk contains candidates[0].content.parts[0].text. The stream termination signal is an empty response or connection close—no explicit sentinel.
• Llama 4 (served via vLLM or TGI) follows the OpenAI-compatible format when using the /v1/chat/completions endpoint, but raw inference endpoints return newline-delimited JSON (NDJSON), not SSE.

These differences mean a single parse_sse_stream() function must handle variable framing, and your assertions must be provider-aware. The following diagram illustrates the test harness architecture that abstracts these concerns:

StreamingTestClient fires POST requests against a FastAPI streaming endpoint, capturing raw SSE byte streams that parse_sse_stream converts into typed ParsedChunk objects. A Provider Router then dispatches each chunk to provider-specific assertion modules—OpenAI, Anthropic, Gemini, and Llama 4/vLLM—before rolling results into a Unified Test Verdict. This architecture validates that every LLM provider's streaming contract holds before the CI/CD Pipeline Gate promotes a build, catching format regressions across four distinct SSE schemas in a single test pass.

Loading diagram...

This architecture separates transport parsing (the parse_sse_stream layer) from semantic validation (provider-specific assertion sets), allowing you to add new providers without rewriting core test infrastructure.

Building the StreamingTestClient and SSE Parser

The foundation of the harness is a StreamingTestClient class that wraps httpx.AsyncClient to issue streaming requests against your FastAPI application and a parse_sse_stream() coroutine generator that consumes raw async byte iterators, reassembles SSE frames, and yields structured ParsedChunk dataclass instances. The client must handle connection-level timeouts separately from per-chunk read timeouts—a critical distinction because a stream that starts successfully but stalls mid-generation should fail differently than one that never connects. The ParsedChunk dataclass normalizes the heterogeneous provider payloads into a uniform shape with event_type, content, raw_data, is_done, and latency_ms fields, enabling downstream assertions to operate provider-agnostically where possible.

1import httpx
2import json
3import time
4from dataclasses import dataclass, field
5from typing import AsyncIterator, Optional
6
7@dataclass
8class ParsedChunk:
9    event_type: str = "message"
10    content: str = ""
11    raw_data: dict = field(default_factory=dict)
12    is_done: bool = False
13    latency_ms: float = 0.0
14
15async def parse_sse_stream(
16    byte_stream: AsyncIterator[bytes],
17    provider: str = "openai",
18) -> AsyncIterator[ParsedChunk]:
19    buffer = ""
20    current_event = "message"
21    stream_start = time.perf_counter()
22
23    async for raw_bytes in byte_stream:
24        buffer += raw_bytes.decode("utf-8", errors="replace")
25        while "\n\n" in buffer:
26            frame, buffer = buffer.split("\n\n", 1)
27            chunk = ParsedChunk(
28                latency_ms=(time.perf_counter() - stream_start) * 1000
29            )
30            for line in frame.strip().split("\n"):
31                if line.startswith("event:"):
32                    current_event = line[len("event:"):].strip()
33                    chunk.event_type = current_event
34                elif line.startswith("data:"):
35                    data_str = line[len("data:"):].strip()
36                    if data_str == "[DONE]":
37                        chunk.is_done = True
38                        yield chunk
39                        return
40                    try:
41                        chunk.raw_data = json.loads(data_str)
42                    except json.JSONDecodeError:
43                        chunk.raw_data = {"_raw": data_str}
44
45            if not chunk.is_done and chunk.raw_data:
46                chunk.content = _extract_content(chunk.raw_data, provider)
47                if current_event == "message_stop":
48                    chunk.is_done = True
49                yield chunk
50
51def _extract_content(data: dict, provider: str) -> str:
52    if provider in ("openai", "llama4"):
53        delta = data.get("choices", [{}])[0].get("delta", {})
54        return delta.get("content", "")
55    elif provider == "anthropic":
56        return data.get("delta", {}).get("text", "")
57    elif provider == "gemini":
58        parts = data.get("candidates", [{}])[0].get("content", {}).get("parts", [])
59        return parts[0].get("text", "") if parts else ""
60    return ""

Writing Provider-Parameterized Async Tests

With the parser in place, the next layer is a pytest-based async test suite using pytest.mark.parametrize to run identical test logic across all four providers. The test below validates three critical streaming properties: that at least one non-empty content chunk arrives, that the concatenated output forms coherent text (not garbled bytes), and that the stream terminates with an explicit done signal within a configurable timeout. The StreamingTestClient class wraps httpx.AsyncClient with the application's ASGI transport so tests execute in-process without needing a running server, while anyio provides the async test backend.

1import pytest
2import httpx
3from httpx import ASGITransport
4from app.main import app  # your FastAPI application
5
6class StreamingTestClient:
7    def __init__(self, timeout: float = 30.0):
8        transport = ASGITransport(app=app)
9        self.client = httpx.AsyncClient(transport=transport, base_url="http://test")
10        self.timeout = httpx.Timeout(timeout, connect=5.0)
11
12    async def stream_completion(self, provider: str, prompt: str):
13        payload = {"model": f"{provider}/default", "messages": [{"role": "user", "content": prompt}], "stream": True}
14        async with self.client.stream("POST", "/v1/chat/completions", json=payload, timeout=self.timeout) as resp:
15            assert resp.status_code == 200, f"Expected 200, got {resp.status_code}"
16            chunks = []
17            async for chunk in parse_sse_stream(resp.aiter_bytes(), provider):
18                chunks.append(chunk)
19            return chunks
20
21PROVIDERS = ["openai", "anthropic", "gemini", "llama4"]
22
23@pytest.mark.anyio
24@pytest.mark.parametrize("provider", PROVIDERS)
25async def test_stream_delivers_content(provider: str):
26    client = StreamingTestClient(timeout=45.0)
27    chunks = await client.stream_completion(provider, "Say hello in one sentence.")
28    content_chunks = [c for c in chunks if c.content]
29    assert len(content_chunks) > 0, f"{provider}: no content chunks received"
30    full_text = "".join(c.content for c in content_chunks)
31    assert len(full_text) > 5, f"{provider}: response too short: {full_text!r}"
32    assert chunks[-1].is_done is True, f"{provider}: stream did not terminate cleanly"
33
34@pytest.mark.anyio
35@pytest.mark.parametrize("provider", PROVIDERS)
36async def test_first_chunk_latency(provider: str):
37    client = StreamingTestClient(timeout=45.0)
38    chunks = await client.stream_completion(provider, "Say hello.")
39    first_content = next((c for c in chunks if c.content), None)
40    assert first_content is not None, f"{provider}: no content chunk found"
41    assert first_content.latency_ms < 3000, (
42        f"{provider}: TTFT {first_content.latency_ms:.0f}ms exceeds 3s threshold"
43    )

Do's and Don'ts

Do's

1. ✓Do normalize heterogeneous provider payloads into a uniform ParsedChunk dataclass inside parse_sse_stream() — keeping transport parsing separate from provider-specific assertion modules means you can add a fifth provider without rewriting OpenAI, Anthropic, or Gemini assertion logic, and a single pytest run covers all four formats in one pass.
2. ✓Do configure connection-level and per-chunk read timeouts as distinct parameters in StreamingTestClient — a stream that opens a TCP connection and then stalls mid-generation is a different failure mode from one that never connects, and collapsing both into a single timeout causes the harness to misclassify hung streams as connectivity errors.
3. ✓Do write provider-specific termination assertions behind the Provider Router for each of the four distinct stream-end signals — OpenAI's data: [DONE] sentinel, Anthropic's event: message_stop, Gemini's connection-close, and Llama 4/vLLM's NDJSON format each require separate checks, so a router that dispatches ParsedChunk objects to the right assertion module catches format regressions a single generic validator will silently miss.

Don'ts

1. ✗Don't assume OpenAI's data: [DONE] sentinel applies to Anthropic or Gemini streams — Anthropic ends with event: message_stop and emits no [DONE] frame, while Gemini signals termination via connection close; a parser that waits for [DONE] will hang indefinitely on both providers, reproducing exactly the hung browser-tab failure the harness is designed to catch.
2. ✗Don't point the StreamingTestClient at Llama 4's raw inference endpoint expecting SSE frames — raw vLLM and TGI inference routes return NDJSON (newline-delimited JSON), not SSE; only the /v1/chat/completions OpenAI-compatible endpoint sends SSE, so parse_sse_stream() must not be applied to the raw route or it will yield zero ParsedChunk objects and pass vacuously.
3. ✗Don't use live API calls to exercise mid-stream error paths like premature disconnects or dropped connections — fault-injection mocks let the CI gate validate that StreamingTestClient propagates those errors correctly across all four providers without burning tokens, hitting rate limits, or making test results non-deterministic on network variance.