Streaming

Arbytra streams chat completions over Server-Sent Events (SSE). Set stream: true and iterate over chunks as they arrive.

Prerequisites

An Arbytra API key
Python 3.10+ with the OpenAI SDK (pip install openai) or the arbytra SDK (pip install arbytra)
- OR Node.js 18+ with the OpenAI SDK (npm install openai) or @arbytra/sdk (npm install @arbytra/sdk)

Stream responses

Stream a chat completion response:

import os
from openai import OpenAI

client = OpenAI(
    api_key=os.environ["ARBYTRA_API_KEY"],
    base_url="https://api.arbytra.com/v1"
)

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

for chunk in stream:
    if chunk.choices and chunk.choices[0].delta.content:
        print(chunk.choices[0].delta.content, end="", flush=True)

import OpenAI from "openai";

const client = new OpenAI({
    apiKey: process.env.ARBYTRA_API_KEY,
    baseURL: "https://api.arbytra.com/v1",
});

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

let content = "";
for await (const chunk of stream) {
    if (chunk.choices[0]?.delta?.content) {
        content += chunk.choices[0].delta.content;
    }
}

console.log(content);

import os
from arbytra import Client

client = Client(
    api_key=os.environ["ARBYTRA_API_KEY"],
    base_url="https://api.arbytra.com/v1"
)

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

for chunk in stream:
    if chunk.choices and chunk.choices[0].delta.content:
        print(chunk.choices[0].delta.content, end="", flush=True)

import { Client } from "@arbytra/sdk";

const client = new Client({
    apiKey: process.env.ARBYTRA_API_KEY,
    baseUrl: "https://api.arbytra.com/v1",
});

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

for await (const chunk of stream) {
    if (chunk.choices[0]?.delta?.content) {
        process.stdout.write(chunk.choices[0].delta.content);
    }
}

curl https://api.arbytra.com/v1/chat/completions \
  -H "Authorization: Bearer $ARBYTRA_API_KEY" \
  -H "Content-Type: application/json" \
  -d '{
    "model": "gpt-4o",
    "messages": [{"role": "user", "content": "Write a short story"}],
    "stream": true
  }'

Stream asynchronously

Stream with the async client:

import os
from openai import AsyncOpenAI
import asyncio

async def stream_response():
    client = AsyncOpenAI(
        api_key=os.environ["ARBYTRA_API_KEY"],
        base_url="https://api.arbytra.com/v1"
    )

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

    async for chunk in stream:
        if chunk.choices and chunk.choices[0].delta.content:
            print(chunk.choices[0].delta.content, end="", flush=True)

asyncio.run(stream_response())

import OpenAI from "openai";

const client = new OpenAI({
    apiKey: process.env.ARBYTRA_API_KEY,
    baseURL: "https://api.arbytra.com/v1",
});

async function streamChat(userMessage: string): Promise<string> {
    const stream = await client.chat.completions.create({
        model: "gpt-4o",
        messages: [{ role: "user", content: userMessage }],
        stream: true,
    });

    let content = "";
    for await (const chunk of stream) {
        if (chunk.choices[0]?.delta?.content) {
            content += chunk.choices[0].delta.content;
        }
    }
    return content;
}

const result = await streamChat("Hello!");
console.log(result);

import os
from arbytra import AsyncClient
import asyncio

async def stream_response():
    client = AsyncClient(
        api_key=os.environ["ARBYTRA_API_KEY"],
        base_url="https://api.arbytra.com/v1"
    )

    stream = await client.chat.completions.create(
        model="gpt-4o",
        messages=[{"role": "user", "content": "Write a short story"}],
        stream=True
    )
    
    async for chunk in stream:
        if chunk.choices and chunk.choices[0].delta.content:
            print(chunk.choices[0].delta.content, end="", flush=True)

asyncio.run(stream_response())

import { Client } from "@arbytra/sdk";

const client = new Client({
    apiKey: process.env.ARBYTRA_API_KEY,
    baseUrl: "https://api.arbytra.com/v1",
});

async function streamResponse() {
    const stream = await client.chat.completions.create({
        model: "gpt-4o",
        messages: [{ role: "user", content: "Write a short story" }],
        stream: true,
    });

    for await (const chunk of stream) {
        if (chunk.choices[0]?.delta?.content) {
            process.stdout.write(chunk.choices[0].delta.content);
        }
    }
}

streamResponse();

Stream events

Each chunk contains:

# ChatCompletionChunk
chunk.id           # "chatcmpl-abc123"
chunk.model        # "gpt-4o"
chunk.created      # 1234567890
chunk.choices[0].delta.content                  # Token content (may be None)
chunk.choices[0].delta.role                     # "assistant" (first chunk only)
chunk.choices[0].delta.reasoning_content        # Reasoning fragment (if model supports it)
chunk.choices[0].delta.reasoning_signature      # Signature for current thinking block
chunk.choices[0].delta.reasoning_redacted_data  # Encrypted redacted thinking data
chunk.choices[0].finish_reason                  # None until last chunk ("stop")
chunk.choices[0].native_finish_reason           # Provider's original value (e.g. "end_turn")

Handle final chunks

The last content chunk carries finish_reason. A trailing chunk carries usage on every stream. You don’t need to set stream_options.

stream = client.chat.completions.create(
    model="gpt-4o",
    messages=[{"role": "user", "content": "Hello!"}],
    stream=True
)

full_content = ""
usage = None

for chunk in stream:
    if chunk.choices:
        if chunk.choices[0].delta.content:
            full_content += chunk.choices[0].delta.content
        if chunk.choices[0].finish_reason:
            print(f"\n\nFinished: {chunk.choices[0].finish_reason}")
    if chunk.usage:
        usage = chunk.usage

if usage:
    print(f"Tokens used: {usage.total_tokens}")

import OpenAI from "openai";

const client = new OpenAI({
    apiKey: process.env.ARBYTRA_API_KEY,
    baseURL: "https://api.arbytra.com/v1",
});

const stream = await client.chat.completions.create({
    model: "gpt-4o",
    messages: [{ role: "user", content: "Hello!" }],
    stream: true,
    stream_options: { include_usage: true },
});

let content = "";
let usage = null;
for await (const chunk of stream) {
    if (chunk.choices[0]?.delta?.content) {
        content += chunk.choices[0].delta.content;
    }
    if (chunk.usage) {
        usage = chunk.usage;
    }
}

console.log(content);
if (usage) console.log(`Tokens: ${usage.total_tokens}`);

stream = client.chat.completions.create(
    model="gpt-4o",
    messages=[{"role": "user", "content": "Hello!"}],
    stream=True
)

full_content = ""
usage = None

for chunk in stream:
    if chunk.choices:
        if chunk.choices[0].delta.content:
            full_content += chunk.choices[0].delta.content
        if chunk.choices[0].finish_reason:
            print(f"\n\nFinished: {chunk.choices[0].finish_reason}")
    if chunk.usage:
        usage = chunk.usage

if usage:
    print(f"Tokens used: {usage.total_tokens}")

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

let fullContent = "";
for await (const chunk of stream) {
    if (chunk.choices[0]?.delta?.content) {
        fullContent += chunk.choices[0].delta.content;
    }
    if (chunk.choices[0]?.finish_reason) {
        console.log(`\n\nFinished: ${chunk.choices[0].finish_reason}`);
    }
}

if (stream.usage) {
    console.log(`Tokens used: ${stream.usage.total_tokens}`);
}

The final streaming chunk always contains token usage. Setting stream_options.include_usage explicitly is harmless but unnecessary.

Stream properties

The stream object exposes usage, routing metadata, and response headers after iteration completes.

Property	Python	TypeScript	Available
Token usage	`stream.usage`	`stream.usage`	After iteration
Routing info	`stream.routing_metadata`	`stream.routing_metadata`	After iteration
Response headers	`stream.response_headers`	`stream.responseHeaders`	Immediately
Close connection	`stream.close()`	`stream.close()`	Any time

Use the stream as a context manager to ensure the connection is released:

with client.chat.completions.create(
    model="gpt-4o",
    messages=[{"role": "user", "content": "Hello!"}],
    stream=True
) as stream:
    for chunk in stream:
        if chunk.choices and chunk.choices[0].delta.content:
            print(chunk.choices[0].delta.content, end="", flush=True)

# Available after iteration
if stream.usage:
    print(f"Tokens: {stream.usage.total_tokens}")
if stream.routing_metadata:
    print(f"Provider: {stream.routing_metadata.provider}")

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

for await (const chunk of stream) {
    if (chunk.choices[0]?.delta?.content) {
        process.stdout.write(chunk.choices[0].delta.content);
    }
}

// Available after iteration
console.log(`Tokens: ${stream.usage?.total_tokens}`);
console.log(`Provider: ${stream.routing_metadata?.provider}`);

Use a context manager for automatic cleanup:

stream = await client.chat.completions.create(
    model="gpt-4o",
    messages=[{"role": "user", "content": "Hello!"}],
    stream=True
)

try:
    async for chunk in stream:
        if chunk.choices and chunk.choices[0].delta.content:
            print(chunk.choices[0].delta.content, end="", flush=True)
finally:
    await stream.close()

import OpenAI from "openai";

const client = new OpenAI({
    apiKey: process.env.ARBYTRA_API_KEY,
    baseURL: "https://api.arbytra.com/v1",
});

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

let content = "";
for await (const chunk of stream) {
    if (chunk.choices[0]?.delta?.content) {
        content += chunk.choices[0].delta.content;
    }
}

console.log(content);

async with await client.chat.completions.create(
    model="gpt-4o",
    messages=[{"role": "user", "content": "Hello!"}],
    stream=True
) as stream:
    async for chunk in stream:
        if chunk.choices and chunk.choices[0].delta.content:
            print(chunk.choices[0].delta.content, end="", flush=True)

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

try {
    for await (const chunk of stream) {
        if (chunk.choices[0]?.delta?.content) {
            process.stdout.write(chunk.choices[0].delta.content);
        }
    }
} finally {
    stream.close();
}

routing_metadata and usage arrive on separate trailing chunks after all content chunks. Consume the stream to completion to access them.

In TypeScript, you can only iterate a stream once. A second attempt throws an error.

Stream with tools

Reassemble streamed tool call chunks into complete function calls:

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

tool_calls = []
for chunk in stream:
    if not chunk.choices:
        continue
    delta = chunk.choices[0].delta

    if delta.tool_calls:
        for tc in delta.tool_calls:
            if tc.index >= len(tool_calls):
                tool_calls.append({"id": tc.id, "function": {"name": "", "arguments": ""}})
            if tc.function and tc.function.name:
                tool_calls[tc.index]["function"]["name"] += tc.function.name
            if tc.function and tc.function.arguments:
                tool_calls[tc.index]["function"]["arguments"] += tc.function.arguments

print(tool_calls)

import OpenAI from "openai";

const client = new OpenAI({
    apiKey: process.env.ARBYTRA_API_KEY,
    baseURL: "https://api.arbytra.com/v1",
});

const tools: OpenAI.ChatCompletionTool[] = [{
    type: "function",
    function: {
        name: "get_weather",
        description: "Get weather for a city",
        parameters: {
            type: "object",
            properties: { city: { type: "string" } },
            required: ["city"],
        },
    },
}];

const stream = await client.chat.completions.create({
    model: "gpt-4o",
    messages: [{ role: "user", content: "What's the weather in Paris?" }],
    tools,
    stream: true,
});

const toolCalls: Record<number, { name: string; arguments: string }> = {};
for await (const chunk of stream) {
    const delta = chunk.choices[0]?.delta;
    if (delta?.tool_calls) {
        for (const tc of delta.tool_calls) {
            if (!toolCalls[tc.index]) {
                toolCalls[tc.index] = { name: "", arguments: "" };
            }
            if (tc.function?.name) toolCalls[tc.index].name = tc.function.name;
            if (tc.function?.arguments) toolCalls[tc.index].arguments += tc.function.arguments;
        }
    }
}

for (const [, tc] of Object.entries(toolCalls)) {
    console.log(`${tc.name}: ${tc.arguments}`);
}

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

tool_calls = []
for chunk in stream:
    if not chunk.choices:
        continue
    delta = chunk.choices[0].delta
    
    # Handle tool call streaming
    if delta.tool_calls:
        for tc in delta.tool_calls:
            if tc.index >= len(tool_calls):
                tool_calls.append({"id": tc.id, "function": {"name": "", "arguments": ""}})
            if tc.function and tc.function.name:
                tool_calls[tc.index]["function"]["name"] += tc.function.name
            if tc.function and tc.function.arguments:
                tool_calls[tc.index]["function"]["arguments"] += tc.function.arguments

print(tool_calls)

const stream = await client.chat.completions.create({
    model: "gpt-4o",
    messages: [{ role: "user", content: "What's the weather in Paris?" }],
    tools: [{
        type: "function",
        function: {
            name: "get_weather",
            description: "Get weather",
            parameters: { type: "object", properties: { city: { type: "string" } } },
        },
    }],
    stream: true,
});

const toolCalls: Array<{ id: string; function: { name: string; arguments: string } }> = [];
for await (const chunk of stream) {
    if (!chunk.choices.length) continue;
    const delta = chunk.choices[0].delta;

    if (delta.tool_calls) {
        for (const tc of delta.tool_calls) {
            if (tc.index >= toolCalls.length) {
                toolCalls.push({ id: tc.id!, function: { name: "", arguments: "" } });
            }
            if (tc.function?.name) toolCalls[tc.index].function.name += tc.function.name;
            if (tc.function?.arguments) toolCalls[tc.index].function.arguments += tc.function.arguments;
        }
    }
}

console.log(toolCalls);

See Tool Calling Guide for function definitions and multi-turn tool conversations.

Stream with routing options

Pass routing options to a streaming request:

stream = client.chat.completions.create(
    model="gpt-5.4",
    messages=[{"role": "user", "content": "Hello!"}],
    stream=True,
    extra_body={"gateway": {"routing": {
        "optimize": "ttft-focus",
        "max_ttft_ms": 1000,
    }}}
)

for chunk in stream:
    if chunk.choices and chunk.choices[0].delta.content:
        print(chunk.choices[0].delta.content, end="", flush=True)

import OpenAI from "openai";

const client = new OpenAI({
    apiKey: process.env.ARBYTRA_API_KEY,
    baseURL: "https://api.arbytra.com/v1",
});

const stream = await client.chat.completions.create({
    model: "gpt-5.4",
    messages: [{ role: "user", content: "Hello!" }],
    stream: true,
    gateway: { routing: {
        optimize: "ttft-focus",
        max_ttft_ms: 1000,
    } },
} as OpenAI.ChatCompletionCreateParamsStreaming);

let content = "";
for await (const chunk of stream) {
    if (chunk.choices[0]?.delta?.content) {
        content += chunk.choices[0].delta.content;
    }
}

if (!content) throw new Error("No streamed content");
console.log(content);

stream = client.chat.completions.create(
    model="gpt-5.4",
    messages=[{"role": "user", "content": "Hello!"}],
    stream=True,
    gateway={
        "routing": {
            "optimize": "ttft-focus",
            "max_ttft_ms": 1000,
        },
    }
)

for chunk in stream:
    if chunk.choices and chunk.choices[0].delta.content:
        print(chunk.choices[0].delta.content, end="", flush=True)

const stream = await client.chat.completions.create({
    model: "gpt-5.4",
    messages: [{ role: "user", content: "Hello!" }],
    stream: true,
    gateway: {
        routing: {
            optimize: "ttft-focus",
            max_ttft_ms: 1000,
        },
    },
});

for await (const chunk of stream) {
    if (chunk.choices[0]?.delta?.content) {
        process.stdout.write(chunk.choices[0].delta.content);
    }
}

curl https://api.arbytra.com/v1/chat/completions \
  -H "Authorization: Bearer $ARBYTRA_API_KEY" \
  -H "Content-Type: application/json" \
  -d '{
    "model": "gpt-5.4",
    "messages": [{"role": "user", "content": "Hello!"}],
    "stream": true,
    "gateway": {"routing": {"optimize": "ttft-focus", "max_ttft_ms": 1000}}
  }'

Handle stream errors

Catch errors during streaming:

import os
from openai import OpenAI, APIStatusError

client = OpenAI(
    api_key=os.environ["ARBYTRA_API_KEY"],
    base_url="https://api.arbytra.com/v1"
)

try:
    stream = client.chat.completions.create(
        model="gpt-4o",
        messages=[{"role": "user", "content": "Hello!"}],
        stream=True
    )

    for chunk in stream:
        if chunk.choices and chunk.choices[0].delta.content:
            print(chunk.choices[0].delta.content, end="", flush=True)

except APIStatusError as e:
    if e.status_code == 429:
        retry_after = e.response.headers.get("retry-after")
        print(f"Rate limited: retry after {retry_after}s")
    else:
        print(f"API error ({e.status_code}): {e.message}")

import OpenAI from "openai";

const client = new OpenAI({
    apiKey: process.env.ARBYTRA_API_KEY,
    baseURL: "https://api.arbytra.com/v1",
});

try {
    const stream = await client.chat.completions.create({
        model: "gpt-4o",
        messages: [{ role: "user", content: "Hello!" }],
        stream: true,
    });

    let content = "";
    for await (const chunk of stream) {
        if (chunk.choices[0]?.delta?.content) {
            content += chunk.choices[0].delta.content;
        }
    }
    console.log(content);
} catch (error) {
    if (error instanceof OpenAI.APIError) {
        console.error(`API Error ${error.status}: ${error.message}`);
        if (error.status === 429) {
            console.error("Rate limited — retry after backoff");
        }
    } else {
        throw error;
    }
}

import os
from arbytra import Client, APIStatusError, RateLimitError

client = Client(
    api_key=os.environ["ARBYTRA_API_KEY"],
    base_url="https://api.arbytra.com/v1"
)

try:
    stream = client.chat.completions.create(
        model="gpt-4o",
        messages=[{"role": "user", "content": "Hello!"}],
        stream=True
    )

    for chunk in stream:
        if chunk.choices and chunk.choices[0].delta.content:
            print(chunk.choices[0].delta.content, end="", flush=True)

except RateLimitError as e:
    print(f"Rate limited: retry after {e.retry_after_seconds}s")
except APIStatusError as e:
    # Mid-stream upstream error (502 upstream_error, 504 upstream_timeout, 503 service_unavailable).
    print(f"Upstream/api error ({e.status_code}, code={e.code}): {e.message}")

import { Client, APIStatusError, RateLimitError } from "@arbytra/sdk";

const client = new Client({
    apiKey: process.env.ARBYTRA_API_KEY,
    baseUrl: "https://api.arbytra.com/v1",
});

try {
    const stream = await client.chat.completions.create({
        model: "gpt-4o",
        messages: [{ role: "user", content: "Hello!" }],
        stream: true,
    });

    for await (const chunk of stream) {
        if (chunk.choices[0]?.delta?.content) {
            process.stdout.write(chunk.choices[0].delta.content);
        }
    }
} catch (e) {
    if (e instanceof RateLimitError) {
        console.log(`Rate limited: retry after ${e.retryAfterSeconds}s`);
    } else if (e instanceof APIStatusError) {
        console.log(`Upstream/api error (${e.statusCode}, code=${e.code}): ${e.message}`);
    }
}

See Error Handling Guide for retry strategies and circuit breakers.

SSE format

Raw SSE events look like this. The stream ends with usage and routing_metadata events before [DONE].

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

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

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

data: {"id":"chatcmpl-123","object":"chat.completion.chunk","created":1234567890,"model":"gpt-4o","choices":[],"usage":{"prompt_tokens":8,"completion_tokens":2,"total_tokens":10}}

data: {"id":"chatcmpl-a1b2c3d4","object":"chat.completion.chunk","created":1234567890,"model":"gpt-4o","choices":[],"routing_metadata":{"provider":"openai","routing_strategy":"balanced","cost":{"usd":0.00015}}}

data: [DONE]

The trailing events before [DONE] carry usage and routing_metadata with choices: []. SDKs expose these as stream.usage and stream.routing_metadata after iteration.

Streams may carry SSE comment lines (for example : ARBYTRA PROCESSING) during quiet phases, such as a reasoning model’s thinking phase. Comment lines start with a colon; spec-compliant SSE parsers skip them.

Get Started

Guides

Response API

Frameworks

Integrations

Specifications

Platform

Changelog

Prerequisites

Stream responses

Stream asynchronously

Stream events

Handle final chunks

Stream properties

Stream with tools

Stream with routing options

Handle stream errors

SSE format

​Prerequisites

​Stream responses

​Stream asynchronously

​Stream events

​Handle final chunks

​Stream properties

​Stream with tools

​Stream with routing options

​Handle stream errors

​SSE format

Prerequisites

Stream responses

Stream asynchronously

Stream events

Handle final chunks

Stream properties

Stream with tools

Stream with routing options

Handle stream errors

SSE format