Streaming

Stream LLM responses through Solwyn — works transparently with all five native providers plus OpenAI-compatible endpoints

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 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 calls. 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.

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.

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 endpoints put it on a final usage chunk, 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() 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.

On this page