Streaming

import os
from openai import OpenAI
from solwyn import Solwyn

with Solwyn(
    OpenAI(),
    api_key=os.environ["SOLWYN_API_KEY"],
) as client:
    stream = client.chat.completions.create(
        model="gpt-4o",
        messages=[{"role": "user", "content": "Write a haiku about latency."}],
        stream=True,
    )

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

Streaming Just Works. Pass stream=True exactly as you would without Solwyn — your code does not change. The OpenAI responses.stream(...) helper is a context manager rather than a flag, and it is metered too; see OpenAI Responses streaming.

The SDK observes the stream's usage data as it flows to your application and reports it to Solwyn Cloud after the stream completes. The chunks are passed through unchanged.

OpenAI streaming

import os
from openai import OpenAI
from solwyn import Solwyn

with Solwyn(
    OpenAI(),
    api_key=os.environ["SOLWYN_API_KEY"],
) as client:
    stream = client.chat.completions.create(
        model="gpt-4o",
        messages=[{"role": "user", "content": "Hello!"}],
        stream=True,
    )

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

Solwyn automatically requests usage data from OpenAI for streaming Chat Completions. You do not need to pass stream_options={"include_usage": True} -- the SDK injects it for you. If you provide your own stream_options dict, the SDK preserves your other keys and adds include_usage=True. This applies to chat.completions.create only: the Responses API's own stream_options has no include_usage key, so the SDK never injects stream_options there -- a Responses stream carries its usage on the terminal event instead.

OpenAI Responses streaming

SDK v0.6.0+. Native OpenAI and Azure OpenAI Responses calls are metered, in both streaming shapes:

import os
from openai import OpenAI
from solwyn import Solwyn

with Solwyn(OpenAI(), api_key=os.environ["SOLWYN_API_KEY"]) as client:
    # Shape 1: create(stream=True) -- iterate the events
    for event in client.responses.create(model="gpt-4o", input="Write a haiku about latency.", stream=True):
        if event.type == "response.output_text.delta":
            print(event.delta, end="", flush=True)

    # Shape 2: the stream() helper -- a context manager with until_done()/get_final_response()
    with client.responses.stream(model="gpt-4o", input="Write a haiku about latency.") as stream:
        for event in stream:
            if event.type == "response.output_text.delta":
                print(event.delta, end="", flush=True)
        final = stream.get_final_response()

Usage arrives on the terminal response.completed event, nested under response.usage; the SDK reads it from there and settles once. Three outcomes are possible for the helper:

Sync and async differ in one way worth knowing: the sync helper runs its budget check and reservation eagerly when you call responses.stream(...) and sends the request at __enter__; the async helper defers everything to the first __aenter__, so an async helper that is never entered performs no budget check at all. responses.stream(response_id=...) and starting_after=... — resuming an existing response — are raw pass-through and never metered.

A stream helper whose request fails at __enter__ is classified like a create(stream=True) failure: a request-shaped 4xx refusal (anything but 429) releases the reservation and reports an error event without counting against the provider's circuit breaker, while a connection failure, a 429, or a 5xx does. Details, refusals, and the Azure specifics are on the OpenAI provider page.

Anthropic streaming

import os
from anthropic import Anthropic
from solwyn import Solwyn

with Solwyn(
    Anthropic(),
    api_key=os.environ["SOLWYN_API_KEY"],
) as client:
    stream = client.messages.create(
        model="claude-sonnet-4-5",
        max_tokens=1024,
        messages=[{"role": "user", "content": "Hello!"}],
        stream=True,
    )

    for event in stream:
        if event.type == "content_block_delta":
            print(event.delta.text, end="", flush=True)

Anthropic streams emit usage data by default. No extra configuration is required.

Google Gemini streaming

import os
from google import genai
from solwyn import Solwyn

gc = genai.Client(api_key=os.environ["GEMINI_API_KEY"])

with Solwyn(
    gc,
    api_key=os.environ["SOLWYN_API_KEY"],
) as client:
    stream = client.models.generate_content_stream(
        model="gemini-2.0-flash",
        contents="Write a haiku about latency.",
    )

    for chunk in stream:
        if chunk.text:
            print(chunk.text, end="", flush=True)

Use generate_content_stream (not generate_content) for streaming. Google streams include usage_metadata on most chunks — Solwyn observes the latest values automatically.

Amazon Bedrock streaming

converse_stream preserves the boto3 contract: you get back a dict whose "stream" value is Solwyn's wrapper around the inner event stream. Iterate it exactly as you would with raw boto3:

import os
import boto3
from solwyn import Solwyn

with Solwyn(
    boto3.client("bedrock-runtime", region_name="us-east-1"),
    api_key=os.environ["SOLWYN_API_KEY"],
) as client:
    result = client.converse_stream(
        modelId="amazon.nova-pro-v1:0",
        messages=[{"role": "user", "content": [{"text": "Write a haiku about latency."}]}],
    )

    with result["stream"]:  # settles the budget reservation even on early break
        for event in result["stream"]:
            if "contentBlockDelta" in event:
                print(event["contentBlockDelta"]["delta"].get("text", ""), end="", flush=True)

Token usage settles from the stream's terminal metadata event when the stream is exhausted. If you stop consuming early, call result["stream"].close() (or wrap iteration in with result["stream"]: as above) so the budget reservation settles with the usage observed — this mirrors the close obligation raw boto3 EventStreams already impose. close() settles exactly once and is safe to call repeatedly. Async (aioboto3): iterate with async for and close with await result["stream"].close() or async with result["stream"]:.

A stream that settles at zero tokens after producing real traffic logs an explicit warning — never silently wrong counts. See Amazon Bedrock.

OpenAI-compatible providers

Streaming works the same way as OpenAI (stream=True on chat.completions.create), but usage delivery varies by endpoint. Solwyn injects stream_options={"include_usage": True} only where that is documented-safe for the detected provider. It never injects for xAI, Mistral, Together, Fireworks, Perplexity, OpenRouter, or the generic openai_compatible catch-all — some of those endpoints reject the parameter outright (xAI, Mistral), one deprecates it (OpenRouter), and the rest leave it undocumented; most deliver usage in the final chunk on their own, and anything that ends up usage-less falls back to the flagged estimation tier below. A stream_options you pass explicitly always reaches your configured provider untouched.

Stream usage resolves in three tiers: the standard usage block (the last chunk whose usage parses to non-zero counts), Groq's legacy x_groq.usage final-chunk shape, and finally an explicit length-based estimate flagged token_details.is_estimated=true with a one-time WARNING — loud, never silently zero. See OpenAI-compatible providers for the per-provider table.

Async streaming

AsyncSolwyn supports streaming with async for:

import asyncio
import os
from openai import AsyncOpenAI
from solwyn import AsyncSolwyn

async def main():
    async with AsyncSolwyn(
        AsyncOpenAI(),
        api_key=os.environ["SOLWYN_API_KEY"],
    ) as client:
        stream = await client.chat.completions.create(
            model="gpt-4o",
            messages=[{"role": "user", "content": "Hello!"}],
            stream=True,
        )

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

asyncio.run(main())

Early abort

Solwyn settles its budget reservation when the stream finishes. If your code needs to break out of the loop early, wrap the stream in a with (sync) or async with (async) block so Solwyn can settle with the tokens it observed:

with client.chat.completions.create(
    model="gpt-4o",
    messages=[{"role": "user", "content": "Hello!"}],
    stream=True,
) as stream:
    for chunk in stream:
        delta = chunk.choices[0].delta.content
        if delta and "STOP" in delta:
            break  # Solwyn settles with tokens observed up to this point
        print(delta or "", end="", flush=True)

For async streams:

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 you fully consume the stream (run to completion), Solwyn settles automatically — no with needed. The with form only matters if you might exit early.

For Bedrock the same obligation applies to the stream inside the contract dict: close (or with-wrap) result["stream"], not the dict itself — see Amazon Bedrock streaming above.

When the run is stopped mid-stream

SDK v0.6.0+. A stream open inside a solwyn.run(...) scope is also aborted from outside when the run is stopped — by an operator on the dashboard or by local velocity detection. The stream is not cut mid-chunk: at the next chunk the SDK pulls from the provider, it discards that chunk, closes the provider stream, settles once with the usage observed so far as a partial success, and raises RunStoppedError from the iteration. That applies to every provider's stream wrapper, the Responses helper, and Bedrock's inner event stream. The stream is terminal afterwards; iterating it again re-raises the same RunStoppedError, and closing it again is a no-op.

from solwyn import RunStoppedError

with solwyn.run("nightly-report"):
    with client.chat.completions.create(model="gpt-4o", messages=[...], stream=True) as stream:
        try:
            for chunk in stream:
                ...
        except RunStoppedError:
            ...   # partial usage is already settled; nothing more to close

What gets reported

After the stream completes (or aborts via with), Solwyn reports:

• input_tokens / output_tokens — extracted from wherever the provider's stream carries usage: OpenAI and compatible Chat Completions endpoints put it on a final usage chunk, OpenAI Responses streams nest it under the terminal response.completed event, Anthropic splits it across the message_start (input) and message_delta (output) events, Google attaches usage_metadata to chunks, and Bedrock delivers it in the terminal metadata event
• latency_ms — elapsed time from create() (or from entering a responses.stream() helper) to stream exhaustion
• All other MetadataEvent fields — the same as a non-streaming call

Token counts and latency are accurate even when you abort early — Solwyn reports what it observed.

Related

• Reference: Solwyn
• Async Usage
• Run control — how a stop reaches an open stream
• OpenAI: Responses API — refusals, Azure, and settlement rules for Responses calls