> ## Documentation Index
> Fetch the complete documentation index at: https://docs.mavera.io/llms.txt
> Use this file to discover all available pages before exploring further.

# Streaming

> Send tokens to users as they're generated instead of waiting for the full response

With a standard API call, you wait until the entire response is generated before anything comes back. That's fine for background jobs, but your users are staring at a blank screen the whole time.

Streaming flips this around. You get tokens the moment they're produced, so your users see the response being typed out in real time — just like ChatGPT. Under the hood, Mavera uses **Server-Sent Events (SSE)** to push each token as it's ready.

## Quick Example

Use `client.responses.stream()` and iterate over events instead of reading a single response.

<CodeGroup>
  ```python Python theme={"dark"}
  import os
  from openai import OpenAI

  client = OpenAI(
      api_key=os.environ["MAVERA_API_KEY"],
      base_url="https://app.mavera.io/api/v1",
  )

  with client.responses.stream(
      model="mavera-1",
      input="Explain API rate limiting in 3 sentences.",
      extra_body={"persona_id": os.environ.get("PERSONA_ID")},
  ) as stream:
      for event in stream:
          if event.type == "response.output_text.delta":
              print(event.delta, end="", flush=True)
  ```

  ```javascript JavaScript theme={"dark"}
  import OpenAI from "openai";

  const client = new OpenAI({
    apiKey: process.env.MAVERA_API_KEY,
    baseURL: "https://app.mavera.io/api/v1",
  });

  const stream = client.responses.stream({
    model: "mavera-1",
    input: "Explain API rate limiting in 3 sentences.",
    persona_id: process.env.PERSONA_ID,
  });

  for await (const event of stream) {
    if (event.type === "response.output_text.delta") {
      process.stdout.write(event.delta);
    }
  }
  ```
</CodeGroup>

That's it. Same client, same model. Use `client.responses.stream()` and handle named events as they arrive.

## How Streaming Works

When you stream, the API doesn't wait to finish generating. Instead, it opens a long-lived HTTP connection and pushes **Server-Sent Events** — one per token (or small group of tokens). Each event has a `type` that tells you what happened. The final `response.completed` event signals the stream is done and includes usage data.

```mermaid theme={"dark"}
sequenceDiagram
    participant App as Your App
    participant API as Mavera API

    App->>API: POST /responses (stream: true)
    API-->>App: response.created
    API-->>App: response.output_item.added
    API-->>App: response.output_text.delta — "Rate"
    API-->>App: response.output_text.delta — " limiting"
    API-->>App: response.output_text.delta — " is"
    API-->>App: response.output_text.delta — "..."
    API-->>App: response.output_text.done
    API-->>App: response.output_item.done
    API-->>App: response.completed
```

The connection stays open until the model finishes or an error occurs. Your client reads events as they arrive, so there's no polling.

<Info>
  Streaming doesn't change what the model generates — you get the exact same output. It only changes *when* you receive it.
</Info>

## Event Structure

Each SSE event is a named event with a `type` field. Here are the key events you'll encounter:

| Event Type                   | Description                                            |
| ---------------------------- | ------------------------------------------------------ |
| `response.created`           | Response object created — streaming has started        |
| `response.output_item.added` | A new output item (text, function call) has been added |
| `response.output_text.delta` | A text token — read it from `event.delta`              |
| `response.output_text.done`  | Text generation for the current item is complete       |
| `response.output_item.done`  | The current output item is fully complete              |
| `response.completed`         | Response is finished — includes full `usage` data      |

A typical stream looks like this:

```json theme={"dark"}
// response.created
{"type": "response.created", "response": {"id": "resp_abc123", "status": "in_progress"}}

// response.output_text.delta events
{"type": "response.output_text.delta", "delta": "Rate"}
{"type": "response.output_text.delta", "delta": " limiting"}
{"type": "response.output_text.delta", "delta": " prevents"}

// response.completed
{"type": "response.completed", "response": {"id": "resp_abc123", "status": "completed", "usage": {"input_tokens": 18, "output_tokens": 42, "total_tokens": 60}}}
```

<Tip>
  Token usage is included automatically in the `response.completed` event — no extra parameters needed.
</Tip>

## Building a Chat UI

In a real application you need the full response text after streaming finishes — for storing in a database, passing to the next API call, or displaying in a conversation thread. Accumulate deltas as they arrive.

<CodeGroup>
  ```python Python theme={"dark"}
  import os
  from openai import OpenAI

  client = OpenAI(
      api_key=os.environ["MAVERA_API_KEY"],
      base_url="https://app.mavera.io/api/v1",
  )

  def stream_response(input_messages, persona_id):
      chunks = []

      with client.responses.stream(
          model="mavera-1",
          input=input_messages,
          extra_body={"persona_id": persona_id},
      ) as stream:
          for event in stream:
              if event.type == "response.output_text.delta":
                  chunks.append(event.delta)
                  print(event.delta, end="", flush=True)

              if event.type == "response.completed":
                  usage = event.response.usage
                  print(f"\n\n[Tokens: {usage.total_tokens}, Credits: {usage.credits_used}]")

      return "".join(chunks)

  full_response = stream_response(
      input_input=[{"role": "user", "content": "What drives Gen Z brand loyalty?"}],
      persona_id=os.environ.get("PERSONA_ID"),
  )
  ```

  ```javascript JavaScript theme={"dark"}
  import OpenAI from "openai";

  const client = new OpenAI({
    apiKey: process.env.MAVERA_API_KEY,
    baseURL: "https://app.mavera.io/api/v1",
  });

  async function streamResponse(inputMessages, personaId) {
    const stream = client.responses.stream({
      model: "mavera-1",
      input: inputMessages,
      persona_id: personaId,
    });

    const chunks = [];
    for await (const event of stream) {
      if (event.type === "response.output_text.delta") {
        chunks.push(event.delta);
        process.stdout.write(event.delta);
      }

      if (event.type === "response.completed") {
        const usage = event.response.usage;
        console.log(`\n\n[Tokens: ${usage.total_tokens}, Credits: ${usage.credits_used}]`);
      }
    }

    return chunks.join("");
  }

  const fullResponse = await streamResponse(
    [{ role: "user", content: "What drives Gen Z brand loyalty?" }],
    process.env.PERSONA_ID,
  );
  ```
</CodeGroup>

For a web frontend, the same pattern applies — push each `event.delta` into your UI state and let your framework re-render. In React, that looks like appending to a `useState` string inside the loop.

## Streaming with Structured Outputs

Structured outputs work with streaming. The JSON arrives token by token just like plain text. You won't have valid JSON until the stream finishes, so accumulate everything, then parse once at the end.

<CodeGroup>
  ```python Python theme={"dark"}
  with client.responses.stream(
      model="mavera-1",
      input=[
          {"role": "user", "content": "Analyze this headline: 'AI Replaces 50% of Marketing Jobs'"}
      ],
      extra_body={
          "persona_id": os.environ.get("PERSONA_ID"),
          "text": {
              "format": {
                  "type": "json_schema",
                  "json_schema": {
                      "name": "headline_analysis",
                      "strict": True,
                      "schema": {
                          "type": "object",
                          "properties": {
                              "sentiment": {"type": "string", "enum": ["positive", "neutral", "negative"]},
                              "clickbait_score": {"type": "number"},
                              "key_claims": {"type": "array", "items": {"type": "string"}},
                              "suggested_revision": {"type": "string"}
                          },
                          "required": ["sentiment", "clickbait_score", "key_claims", "suggested_revision"]
                      }
                  }
              }
          }
      },
  ) as stream:
      raw = []
      for event in stream:
          if event.type == "response.output_text.delta":
              raw.append(event.delta)
              print(event.delta, end="", flush=True)

  result = json.loads("".join(raw))
  print(f"\n\nSentiment: {result['sentiment']}")
  print(f"Clickbait score: {result['clickbait_score']}")
  ```

  ```javascript JavaScript theme={"dark"}
  const stream = client.responses.stream({
    model: "mavera-1",
    input: [
      { role: "user", content: "Analyze this headline: 'AI Replaces 50% of Marketing Jobs'" },
    ],
    persona_id: process.env.PERSONA_ID,
    text: {
      format: {
        type: "json_schema",
        json_schema: {
          name: "headline_analysis",
          strict: true,
          schema: {
            type: "object",
            properties: {
              sentiment: { type: "string", enum: ["positive", "neutral", "negative"] },
              clickbait_score: { type: "number" },
              key_claims: { type: "array", items: { type: "string" } },
              suggested_revision: { type: "string" },
            },
            required: ["sentiment", "clickbait_score", "key_claims", "suggested_revision"],
          },
        },
      },
    },
  });

  const raw = [];
  for await (const event of stream) {
    if (event.type === "response.output_text.delta") {
      raw.push(event.delta);
      process.stdout.write(event.delta);
    }
  }

  const result = JSON.parse(raw.join(""));
  console.log(`\nSentiment: ${result.sentiment}`);
  console.log(`Clickbait score: ${result.clickbait_score}`);
  ```
</CodeGroup>

<Info>
  You can show a live JSON preview while streaming by attempting `JSON.parse()` on each accumulated chunk. Libraries like [`partial-json`](https://www.npmjs.com/package/partial-json) can parse incomplete JSON for real-time UI updates.
</Info>

## Streaming with Function Calling

When the model decides to call a tool, the function name and arguments stream in as events. You'll receive `response.function_call_arguments.delta` events with argument fragments. Accumulate them the same way you accumulate text content.

<CodeGroup>
  ```python Python theme={"dark"}
  tools = [{
      "type": "function",
      "name": "lookup_competitor",
      "description": "Look up competitor information by name",
      "parameters": {
          "type": "object",
          "properties": {
              "company": {"type": "string", "description": "Competitor company name"},
              "metrics": {
                  "type": "array",
                  "items": {"type": "string"},
                  "description": "Metrics to retrieve"
              }
          },
          "required": ["company"]
      }
  }]

  with client.responses.stream(
      model="mavera-1",
      input=[{"role": "user", "content": "Compare our pricing to Acme Corp"}],
      extra_body={"persona_id": os.environ.get("PERSONA_ID"), "tools": tools},
  ) as stream:
      function_name = ""
      call_id = ""
      arguments = ""

      for event in stream:
          if event.type == "response.output_item.added":
              if hasattr(event, "item") and event.item.type == "function_call":
                  function_name = event.item.name
                  call_id = event.item.call_id

          if event.type == "response.function_call_arguments.delta":
              arguments += event.delta

          if event.type == "response.output_item.done":
              if hasattr(event, "item") and event.item.type == "function_call":
                  print(f"Function: {function_name}")
                  parsed_args = json.loads(arguments)
  ```

  ```javascript JavaScript theme={"dark"}
  const tools = [{
    type: "function",
    name: "lookup_competitor",
    description: "Look up competitor information by name",
    parameters: {
      type: "object",
      properties: {
        company: { type: "string", description: "Competitor company name" },
        metrics: { type: "array", items: { type: "string" }, description: "Metrics to retrieve" },
      },
      required: ["company"],
    },
  }];

  const stream = client.responses.stream({
    model: "mavera-1",
    input: [{ role: "user", content: "Compare our pricing to Acme Corp" }],
    persona_id: process.env.PERSONA_ID,
    tools,
  });

  let functionName = "";
  let callId = "";
  let args = "";

  for await (const event of stream) {
    if (event.type === "response.output_item.added" && event.item?.type === "function_call") {
      functionName = event.item.name;
      callId = event.item.call_id;
    }

    if (event.type === "response.function_call_arguments.delta") {
      args += event.delta;
    }

    if (event.type === "response.output_item.done" && event.item?.type === "function_call") {
      console.log(`Function: ${functionName}`);
      console.log(`Arguments: ${args}`);
      const parsedArgs = JSON.parse(args);
    }
  }
  ```
</CodeGroup>

After you receive the full tool call, execute the function locally and send the result back in a follow-up request (streaming again if you want the final answer streamed too).

## Error Handling

Streams can fail mid-way. A network hiccup, a server timeout, or a client disconnect can leave you with a partial response. Here's how to handle the common cases.

### Connection Drops and Timeouts

Wrap your stream in a try/catch to handle broken connections gracefully. Decide whether to retry (if idempotent) or surface the partial response to the user.

<CodeGroup>
  ```python Python theme={"dark"}
  from openai import APIConnectionError, APITimeoutError

  def stream_with_recovery(input_data, persona_id, max_retries=3):
      for attempt in range(max_retries):
          chunks = []
          try:
              with client.responses.stream(
                  model="mavera-1",
                  input=input_data,
                  extra_body={"persona_id": persona_id},
              ) as stream:
                  for event in stream:
                      if event.type == "response.output_text.delta":
                          chunks.append(event.delta)
                          print(event.delta, end="", flush=True)

              return "".join(chunks)

          except (APIConnectionError, APITimeoutError) as e:
              partial = "".join(chunks)
              print(f"\n\n[Connection lost after {len(partial)} chars. Attempt {attempt + 1}/{max_retries}]")

              if attempt == max_retries - 1:
                  if partial:
                      print("[Returning partial response]")
                      return partial
                  raise

          except Exception as e:
              print(f"\n[Stream error: {e}]")
              raise
  ```

  ```javascript JavaScript theme={"dark"}
  async function streamWithRecovery(input, personaId, maxRetries = 3) {
    for (let attempt = 0; attempt < maxRetries; attempt++) {
      const chunks = [];
      try {
        const stream = client.responses.stream({
          model: "mavera-1",
          input,
          persona_id: personaId,
        });

        for await (const event of stream) {
          if (event.type === "response.output_text.delta") {
            chunks.push(event.delta);
            process.stdout.write(event.delta);
          }
        }

        return chunks.join("");
      } catch (error) {
        const partial = chunks.join("");
        console.error(`\n[Connection lost after ${partial.length} chars. Attempt ${attempt + 1}/${maxRetries}]`);

        if (attempt === maxRetries - 1) {
          if (partial) return partial;
          throw error;
        }
      }
    }
  }
  ```
</CodeGroup>

### Checklist

<AccordionGroup>
  <Accordion title="Set a timeout on the client">
    The OpenAI SDK lets you pass `timeout` (in seconds for Python, milliseconds for JS). Without a timeout, a stalled connection can hang forever. 60 seconds is a reasonable default.
  </Accordion>

  <Accordion title="Watch for empty streams">
    If the very first event errors out, you'll get an exception before any content arrives. Handle this the same as a non-streaming API error — retry or surface the error to the user.
  </Accordion>

  <Accordion title="Handle partial JSON in structured outputs">
    If the stream drops while returning JSON, you'll have invalid JSON. Don't try to parse it — surface a user-friendly error and retry the request.
  </Accordion>

  <Accordion title="Rate limit errors still return 429">
    If you're rate-limited, the streaming request fails before any events are sent. You'll get a `RateLimitError` (Python) or a response with `status: 429` (JS). Handle this with exponential backoff, same as non-streaming.
  </Accordion>
</AccordionGroup>

<Warning>
  Streaming requests consume the same credits as standard requests. A dropped connection still costs credits for the tokens that were generated before the disconnect.
</Warning>

## When to Use Streaming

Streaming isn't always the right choice. Here's a quick decision guide:

| Use Case                              | Streaming                                       | Standard               |
| ------------------------------------- | ----------------------------------------------- | ---------------------- |
| Chat UIs and conversational apps      | **Yes** — users see responses instantly         | No — awkward delay     |
| Long-form content (articles, reports) | **Yes** — show progress on long generations     | Depends on context     |
| Batch processing and pipelines        | No — overhead of event handling isn't worth it  | **Yes** — simpler code |
| Structured outputs (JSON)             | Either — stream for UX, standard for simplicity | Either                 |
| Function calling                      | Either — stream to show "thinking" state        | Either                 |
| Webhooks and async workflows          | No — you need the full response in one payload  | **Yes**                |

<Tip>
  A good rule of thumb: if a human is watching, stream it. If a machine is consuming it, use standard.
</Tip>

## See Also

<CardGroup cols={2}>
  <Card title="Responses API" icon="message" href="/features/responses">
    Full API reference for responses, including all parameters
  </Card>

  <Card title="Structured Outputs" icon="code" href="/features/responses#structured-outputs">
    JSON mode and JSON Schema for typed responses
  </Card>

  <Card title="Error Handling" icon="exclamation-triangle" href="/guides/errors">
    Complete error codes and retry strategies
  </Card>

  <Card title="Rate Limits" icon="gauge" href="/guides/rate-limits">
    Request limits, headers, and backoff patterns
  </Card>
</CardGroup>