OpenAI Responses API Streaming Events Errors — Fix Guide (2026)
Home/ Errors/ OpenAI/ Streaming events errors
Responses API · Streaming Severity: Medium

OpenAI Responses API Streaming Events Errors

Streaming a Responses call gives you a typed event stream — response.output_text.delta for text tokens, response.function_call_arguments.delta for tool calls, plus lifecycle events for every output item. Miss the right event names and your UI shows nothing; assemble them wrong and messages come back scrambled. Here's the event map and the fixes.

By Ahmed R. · Last updated Aug 14, 2026 · OpenAI · Page #139

TL;DRSet stream=True on responses.create and iterate. The stream emits typed events with event.type: consume response.output_text.delta for token chunks, response.function_call_arguments.delta for tool-call arg chunks, and response.completed for the final response object. Older docs reference deprecated names like response.audio.delta and response.text.delta — the GA names use output_text and output_audio prefixes.

Real error messages you'll see

No chunks arriving despite stream=True
No chunks arriving despite stream=True
# for event in stream: emits nothing until the whole response is done
# Root cause: filtering for the wrong event name (e.g. old response.text.delta instead of response.output_text.delta)
# or the client SDK version doesn't recognize the new event names — upgrade.
AttributeError — event.delta not found
AttributeError — event.delta not found
AttributeError: 'ResponseOutputTextDeltaEvent' object has no attribute 'delta'
  at chunk = event.delta
# The delta lives on event.delta in some versions, event.text in others.
# Robust code: check hasattr, or use the SDK's typed accessors.
Tool call arguments arrive as JSON fragments
Tool call arguments arrive as JSON fragments
# Each function_call_arguments.delta event carries a partial JSON string.
# Concatenating them naively before all deltas arrive gives invalid JSON.
# Fix: buffer per item_id and parse only after response.output_item.done for that call.

Responses API stream events (most useful)

Event typeWhat it carries
response.createdResponse object created, id assigned
response.in_progressResponse has started generating
response.output_item.addedA new output item (message, tool call, reasoning) is starting
response.output_text.deltaPartial text token (data.delta or data.text)
response.output_text.doneText output item complete
response.function_call_arguments.deltaPartial JSON arg string for a function call
response.function_call_arguments.doneFunction call args complete (data.arguments = full JSON)
response.output_audio.deltaPartial audio bytes (base64)
response.reasoning_summary_text.deltaReasoning summary chunk (o-series models)
response.output_item.doneAn output item finished (message, tool call, etc.)
response.completedFinal event; carries the full response object
response.failed / response.incompleteTerminal error events

Root causes (ranked by frequency)

Based on OpenAI developer reports; percentages sum to 96%.

  • 21%
    Filtering on deprecated event names. response.text.delta, response.audio.delta, plain message — these are older shapes. Current GA uses response.output_text.delta, response.output_audio.delta.
  • 17%
    Delta accessed as event.delta when it lives on event.text (or vice versa). Field names vary by SDK version. Use hasattr or upgrade to a recent SDK with typed events.
  • 14%
    Function call args JSON parsed too early. Each delta is a JSON fragment. Buffer per item_id until the corresponding done event fires, then parse.
  • 12%
    Multiple output items interleaved. One response may emit message text, a tool call, and reasoning summary in parallel deltas. Not tracking item_id per event conflates them.
  • 10%
    Backpressure — client can't consume fast enough. Consumer buffers fill and the server may terminate the stream. Add rate-limiting or drop UI-only intermediate updates.
  • 8%
    Not handling response.failed. Stream ends without completed; consumer waits forever. Always handle terminal error events.
  • 8%
    Streaming disabled server-side for the model. Some newer models had streaming gated during rollout. Check the model capabilities table.
  • 6%
    Consuming events but not the final response object. The response.completed event carries the full response — including usage, IDs, and metadata. Skipping it loses billing/audit data.

    • How to fix it

    Fix #1

    Iterate the stream, dispatch on event.type, buffer per item_id

    The general shape of a correct stream consumer.

    The stream yields typed events. Dispatch on event.type and route each event to the right handler. For text, append event.delta (or event.text) to a per-item-id buffer. For function calls, buffer the JSON string per item and parse when the done event fires. Always handle response.completed for the final response object.

    stream_consumer.pypython
    from openai import OpenAI
import json

client = OpenAI()


def stream_response(user_input: str, tools: list = None):
    """Consume a Responses API stream and print text as it arrives."""
    text_buffers: dict[str, list[str]] = {}       # item_id → text chunks
    fn_arg_buffers: dict[str, list[str]] = {}     # item_id → JSON arg chunks
    fn_call_meta: dict[str, dict] = {}            # item_id → {name, call_id}

    stream = client.responses.create(
        model="gpt-5.4",
        input=user_input,
        tools=tools or [],
        stream=True,
    )

    for event in stream:
        t = event.type

        if t == "response.created":
            print(f"[response {event.response.id} created]")

        elif t == "response.output_item.added":
            item = event.item
            if item.type == "function_call":
                fn_call_meta[item.id] = {"name": item.name, "call_id": item.call_id}
                fn_arg_buffers[item.id] = []
                print(f"\n[tool call starting: {item.name}]")

        elif t == "response.output_text.delta":
            # Text token chunk — write to stdout, buffer for reconstruction
            delta = getattr(event, "delta", None) or getattr(event, "text", "")
            text_buffers.setdefault(event.item_id, []).append(delta)
            print(delta, end="", flush=True)

        elif t == "response.function_call_arguments.delta":
            # JSON arg fragment — DO NOT parse until done
            delta = getattr(event, "delta", None) or getattr(event, "arguments", "")
            fn_arg_buffers.setdefault(event.item_id, []).append(delta)

        elif t == "response.function_call_arguments.done":
            # Full args now available
            full_args = "".join(fn_arg_buffers.get(event.item_id, []))
            args = json.loads(full_args)
            meta = fn_call_meta.get(event.item_id, {})
            print(f"\n[tool call ready: {meta.get('name')}({args})]")
            # Dispatch to your tool implementation here

        elif t == "response.reasoning_summary_text.delta":
            # o-series reasoning summary chunks
            print(getattr(event, "delta", ""), end="", flush=True)

        elif t == "response.output_item.done":
            # An item finished — good place to finalize
            pass

        elif t == "response.completed":
            resp = event.response
            print(f"\n\n[done — usage: {resp.usage.total_tokens} tokens]")
            return resp

        elif t == "response.failed":
            print(f"\n[FAILED: {event.response.error}]")
            return event.response

        elif t == "response.incomplete":
            print(f"\n[INCOMPLETE: reason={event.response.incomplete_details.reason}]")
            return event.response


# Usage
stream_response("Explain the Responses API stream event model.")


# ✅ Robust delta access — SDK versions vary on field naming
def get_delta(event):
    for attr in ("delta", "text", "arguments", "part"):
        v = getattr(event, attr, None)
        if isinstance(v, str):
            return v
    return ""
    Note: The critical invariant: never parse function call arguments from partial deltas. Each delta may split a JSON string mid-token. Buffer until the done event, then json.loads once. Naive concat-and-parse is the top cause of "tool arguments malformed" bugs.
    Fix #2

    Stream to a browser via FastAPI SSE — forward token deltas as they arrive

    The production pattern for chat UI streaming.

    Wrap the stream in an SSE (Server-Sent Events) response. Forward each output_text.delta as a JSON payload; the browser reconstructs the message. Handle client disconnect by canceling the stream to avoid runaway generations.

    fastapi_sse.pypython
    from fastapi import FastAPI, Request
from fastapi.responses import StreamingResponse
from openai import AsyncOpenAI
import json
import asyncio

api = FastAPI()
client = AsyncOpenAI()


async def sse_from_responses(
    user_msg: str,
    conversation_id: str | None,
    request: Request,
):
    """Yield SSE messages from a Responses API stream."""
    fn_arg_buffers: dict[str, list[str]] = {}

    try:
        stream = await client.responses.create(
            model="gpt-5.4",
            input=user_msg,
            conversation=conversation_id,
            tools=[
                {"type": "web_search"},
            ],
            stream=True,
        )

        async for event in stream:
            # Cancel if client disconnected
            if await request.is_disconnected():
                await stream.close()
                break

            t = event.type

            if t == "response.output_text.delta":
                delta = getattr(event, "delta", None) or getattr(event, "text", "")
                yield f"data: {json.dumps({'type': 'text', 'chunk': delta})}\n\n"

            elif t == "response.function_call_arguments.delta":
                # Buffer server-side, don't stream JSON fragments to the browser
                fn_arg_buffers.setdefault(event.item_id, []).append(
                    getattr(event, "delta", "") or ""
                )

            elif t == "response.function_call_arguments.done":
                full = "".join(fn_arg_buffers.get(event.item_id, []))
                # Send the completed tool call to the browser (e.g. show "searching web...")
                yield f"data: {json.dumps({'type': 'tool_call', 'args_json': full})}\n\n"

            elif t == "response.output_item.added":
                item = event.item
                if item.type == "web_search_call":
                    yield f"data: {json.dumps({'type': 'searching'})}\n\n"

            elif t == "response.completed":
                yield f"data: {json.dumps({'type': 'done', 'response_id': event.response.id})}\n\n"

            elif t == "response.failed":
                err = str(event.response.error) if event.response.error else "unknown"
                yield f"data: {json.dumps({'type': 'error', 'message': err})}\n\n"

    except Exception as e:
        yield f"data: {json.dumps({'type': 'error', 'message': str(e)})}\n\n"


@api.post("/chat/stream")
async def chat_stream(request: Request, message: str, conversation_id: str | None = None):
    return StreamingResponse(
        sse_from_responses(message, conversation_id, request),
        media_type="text/event-stream",
        headers={"Cache-Control": "no-cache", "X-Accel-Buffering": "no"},
    )


# ✅ Coalesce chunks to avoid too many SSE messages under fast streams
async def coalesced_sse(events, flush_ms: int = 40):
    """Buffer deltas for up to flush_ms, emit joined content."""
    buffer: list[str] = []
    import time
    last_flush = time.perf_counter() * 1000

    async for event in events:
        if event.type == "response.output_text.delta":
            delta = getattr(event, "delta", None) or getattr(event, "text", "")
            buffer.append(delta)
            now = time.perf_counter() * 1000
            if now - last_flush >= flush_ms:
                yield f"data: {json.dumps({'type': 'text', 'chunk': ''.join(buffer)})}\n\n"
                buffer.clear()
                last_flush = now
        elif event.type == "response.completed":
            if buffer:
                yield f"data: {json.dumps({'type': 'text', 'chunk': ''.join(buffer)})}\n\n"
            yield f"data: {json.dumps({'type': 'done'})}\n\n"
    Note: The X-Accel-Buffering: no header disables nginx buffering — without it, SSE messages queue at the proxy and users see stalls. Add it whenever you stream through nginx or similar reverse proxies.
    Fix #3

    Handle o-series reasoning summaries and audio streams distinctly

    Fixes missing reasoning UI and audio playback bugs.

    Two special stream types deserve their own handlers. o-series models emit response.reasoning_summary_text.delta events — treat these as a separate UI channel (thinking pane, collapsed detail). Realtime and audio-out responses emit response.output_audio.delta with base64-encoded audio bytes — buffer and feed to your audio player.

    reasoning_and_audio.pypython
    from openai import OpenAI
import base64

client = OpenAI()


# ✅ o-series reasoning — separate reasoning stream from answer stream
def stream_with_reasoning(question: str):
    stream = client.responses.create(
        model="o3",
        input=question,
        reasoning={"effort": "medium", "summary": "auto"},   # request reasoning summary
        stream=True,
    )

    reasoning_chunks: list[str] = []
    answer_chunks: list[str] = []

    for event in stream:
        t = event.type
        if t == "response.reasoning_summary_text.delta":
            # Stream reasoning to a separate UI channel
            chunk = getattr(event, "delta", "")
            reasoning_chunks.append(chunk)
            print(f"[think] {chunk}", end="", flush=True)

        elif t == "response.output_text.delta":
            # Final answer text
            chunk = getattr(event, "delta", None) or getattr(event, "text", "")
            answer_chunks.append(chunk)
            print(chunk, end="", flush=True)

        elif t == "response.completed":
            usage = event.response.usage
            print(f"\n[reasoning_tokens: {usage.output_tokens_details.reasoning_tokens}]")

    return {
        "reasoning": "".join(reasoning_chunks),
        "answer":    "".join(answer_chunks),
    }


# ✅ Audio streaming — buffer base64 chunks and decode when complete
def stream_audio_response(prompt: str):
    stream = client.responses.create(
        model="gpt-5.4",                        # or an audio-out model
        input=prompt,
        modalities=["text", "audio"],
        audio={"voice": "cedar", "format": "pcm16"},
        stream=True,
    )

    audio_chunks: list[bytes] = []
    for event in stream:
        if event.type == "response.output_audio.delta":
            # delta is base64-encoded audio bytes
            b64 = getattr(event, "delta", "") or ""
            if b64:
                audio_chunks.append(base64.b64decode(b64))
        elif event.type == "response.output_audio_transcript.delta":
            # Transcript of what's being spoken — good for captions
            print(getattr(event, "delta", ""), end="", flush=True)

    full_pcm = b"".join(audio_chunks)
    with open("out.pcm", "wb") as f:
        f.write(full_pcm)
    print(f"\n")


# ✅ Detect terminal events consistently
TERMINAL = {"response.completed", "response.failed", "response.incomplete"}

def is_terminal(event) -> bool:
    return event.type in TERMINAL


# ✅ Timeout safety — kill the stream if no event arrives for N seconds
import asyncio
from openai import AsyncOpenAI

async def stream_with_timeout(prompt: str, per_event_timeout: float = 30.0):
    aclient = AsyncOpenAI()
    stream = await aclient.responses.create(model="gpt-5.4", input=prompt, stream=True)

    async def next_event(it):
        return await anext(it)

    it = stream.__aiter__()
    while True:
        try:
            event = await asyncio.wait_for(next_event(it), timeout=per_event_timeout)
        except asyncio.TimeoutError:
            await stream.close()
            raise TimeoutError(f"No event in {per_event_timeout}s; stream aborted")

        if event.type == "response.output_text.delta":
            print(getattr(event, "delta", ""), end="", flush=True)
        if is_terminal(event):
            break
    Note: Reasoning summary is opt-in via reasoning.summary="auto" (or "detailed"). Without it, o-series models still generate reasoning tokens internally but no summary events fire — you only see the final answer. Always request the summary if your UI has a "show reasoning" affordance.

    Prevention checklist

    • Use current GA event names: response.output_text.delta, response.output_audio.delta, response.function_call_arguments.delta. Deprecated names still exist in older docs.
    • Access delta content via getattr(event, "delta", ...) — field naming varies across SDK versions.
    • Buffer function call argument JSON per item_id. Never parse until the corresponding done event fires.
    • Handle response.completed, response.failed, and response.incomplete — all three are terminal.
    • For SSE proxies, set X-Accel-Buffering: no to prevent nginx-style buffering that stalls streaming.
    • Coalesce text deltas into 30-50ms batches for chat UIs — smoother visually and cheaper on the wire.
    • For o-series models, opt into reasoning summaries via reasoning.summary="auto" and route them to a separate UI channel.

    Frequently asked questions

    Can I get the final response object from a stream?

    Yes — the last event is response.completed and it carries the full response object under event.response. This includes id, usage, output, metadata, and everything else you'd get from a non-streaming call. Always consume this event, even for pure UI streaming — you'll need it for billing, logging, and audit trails.

    Does streaming work for function calls?

    Yes — tool call arguments stream as response.function_call_arguments.delta events. Each delta is a partial JSON string. The full arguments are available after response.function_call_arguments.done, which carries the complete JSON in event.arguments. Never try to parse partial deltas as JSON — split points don't respect token boundaries.

    What happens if the client disconnects mid-stream?

    The server keeps generating until the response completes or times out — you're still billed. To abort, close the stream explicitly: with the SDK, call stream.close(); in FastAPI, check request.is_disconnected() inside your event loop. Add this to your SSE handlers so runaway generations don't rack up token cost after the user closed the tab.

    Can I combine streaming with previous_response_id chains?

    Yes — pass previous_response_id and stream=True together. The stream still emits the full event sequence and the final response.completed carries the new response ID, which you use as the next call's previous_response_id. Same for the Conversations API: pass conversation=... and stream=True together.

    Why does my stream sometimes end with response.incomplete instead of response.completed?

    response.incomplete means the model stopped before finishing — usually because max_output_tokens was reached, a content policy filter fired, or the model detected a safety issue. The event carries event.response.incomplete_details.reason. For length limits, raise max_output_tokens; for policy stops, review the input; for reasoning-token exhaustion (o-series), see error #140.

    Related errors

    Related error
    Chat Completions → Responses API migration
    Related error
    Responses API built-in tools errors
    Related error
    o-series max_completion_tokens exhausted
    Hub · every error
    OpenAI — every error (2026)