How Streaming Works in aisuite Across Different LLM Providers: A Code-Level Breakdown

aisuite unifies LLM streaming behind a single client.chat.completions.create(..., stream=True) call by converting every provider's native stream into a standard ChatCompletionChunk object, automatically supplying an async bridge for providers that only implement synchronous streaming.

Streaming in aisuite is engineered to hide vendor-specific differences behind one consistent Python API. According to the andrewyng/aisuite source code, the library normalizes disparate formats—whether they come from OpenAI, Anthropic, Gemini, or another service—into a single OpenAI-style chunk structure. Every returned object is an instance of aisuite.framework.chat_completion_chunk.ChatCompletionChunk, giving you the same fields regardless of which model is running in the background.

Three-Layer Streaming Architecture

The streaming implementation is split into three layers. The provider base class in aisuite/provider.py defines the interface contract and a generic async fallback. Provider-specific overrides in files such as aisuite/providers/openai_provider.py, aisuite/providers/anthropic_provider.py, and aisuite/providers/gemini_provider.py translate native SDK streams into the common chunk format. The client façade in aisuite/client.py inspects the stream=True flag and delegates to the appropriate provider method.

This design means that adding streaming support for a new provider usually requires implementing only a synchronous generator; the base class handles the async counterpart automatically.

Provider Base Class and the Default Async Bridge

The abstract Provider class in aisuite/provider.py establishes the streaming contract. By default, chat_completions_create_stream raises an LLMError:

def chat_completions_create_stream(self, model, messages, **kwargs):
    raise LLMError(f"{type(self).__name__} does not support streaming chat completions.")

If a provider overrides this sync method but does not define an async equivalent, the base class supplies achat_completions_create_stream. This method runs the synchronous stream in a worker thread through run_in_executor and forwards each chunk through an asyncio.Queue:

async def achat_completions_create_stream(self, model, messages, **kwargs):
    loop = asyncio.get_running_loop()
    queue: asyncio.Queue = asyncio.Queue()

    def produce():
        try:
            for chunk in self.chat_completions_create_stream(model, messages, **kwargs):
                loop.call_soon_threadsafe(queue.put_nowait, ("chunk", chunk))
        except BaseException as exc:
            loop.call_soon_threadsafe(queue.put_nowait, ("error", exc))
        finally:
            loop.call_soon_threadsafe(queue.put_nowait, ("done", None))

    loop.run_in_executor(None, produce)
    while True:
        kind, payload = await queue.get()
        if kind == "chunk":
            yield payload
        elif kind == "error":
            raise payload
        else:
            return

Any provider that implements the sync generator instantly gains a non-blocking async interface without extra code.

How Each LLM Provider Implements Streaming in aisuite

Providers that support streaming natively override chat_completions_create_stream to return their own generators, then normalize the output to aisuite's ChatCompletionChunk.

OpenAI: Direct Pass-Through

aisuite/providers/openai_provider.py returns the OpenAI SDK's native iterator directly because it already emits objects matching aisuite's expected shape. The synchronous and asynchronous implementations are thin wrappers:

def chat_completions_create_stream(self, model, messages, **kwargs):
    stream = self.client.chat.completions.create(
        model=model,
        messages=self.transformer.convert_request(messages),
        stream=True,
        **kwargs,
    )
    yield from stream

async def achat_completions_create_stream(self, model, messages, **kwargs):
    stream = await self.aclient.chat.completions.create(
        model=model,
        messages=self.transformer.convert_request(messages),
        stream=True,
        **kwargs,
    )
    async for chunk in stream:
        yield chunk

No additional conversion is required, so OpenAI streams move straight from the SDK to your loop.

Anthropic: Event-to-Chunk Conversion

In aisuite/providers/anthropic_provider.py—with the streaming logic starting around line 387—the Anthropic SDK returns a stream of mixed-type events. The provider consumes these events and feeds each one through AnthropicMessageConverter.convert_stream_event, defined at lines 62–154:

def chat_completions_create_stream(self, model, messages, **kwargs):
    events = self.client.messages.create(
        model=model,
        system=system_message,
        messages=converted_messages,
        stream=True,
        **kwargs,
    )
    state = {}
    for event in events:
        chunk = self.converter.convert_stream_event(event, state)
        if chunk is not None:
            yield chunk

The converter handles tool-use start events, text deltas, and final message metadata, mapping them into ChatCompletionChunk objects with correct finish_reason and usage fields.

Gemini: Stateful Folding

aisuite/providers/gemini_provider.py (streaming logic begins around line 409) uses a small state machine called _StreamState, found at lines 31–92, to fold Gemini's generate_content_stream chunks into OpenAI-shaped chunks. It also synthesizes tool calls and attaches usage metadata:

def chat_completions_create_stream(self, model, messages, **kwargs):
    request = self._request_kwargs(model, messages, kwargs)
    state = _StreamState()
    yield state.role_chunk()
    for chunk in self.client.models.generate_content_stream(**request):
        for out in state.convert(chunk, self.converter):
            yield out
    yield state.final_chunk(self.converter)

Because Gemini's streaming API delivers whole tool-call parts, _StreamState can emit each call as a single chunk without accumulating incremental deltas across multiple events.

Other Providers: Thread-Based Fallback

Providers such as Mistral, Cohere, and Fireworks do not implement their own chat_completions_create_stream. When you request stream=True with these backends, aisuite falls back to the base class bridge. The synchronous stream runs inside a background thread, while your async code receives chunks via the asyncio.Queue mechanism shown previously. This guarantees a uniform non-blocking interface for every supported vendor.

Practical Streaming Examples with aisuite

Synchronous Streaming with Any Provider

The following snippet works with OpenAI, Anthropic, Gemini, or any other provider that either natively supports streaming or relies on the default bridge:

from aisuite import Client

client = Client(provider_configs={"openai": {"api_key": "sk-…"}})

stream = client.chat.completions.create(
    model="openai:gpt-4o-mini",
    messages=[{"role": "user", "content": "Summarise the plot of Inception"}],
    stream=True,
)

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

Each chunk is a ChatCompletionChunk, so you can consume it the same way regardless of the backend.

Async Streaming with Tool Calls (Anthropic)

Anthropic's async stream is converted automatically. Tool-call chunks appear as delta.tool_calls just as they would in OpenAI's format:

import asyncio
from aisuite import Client

client = Client(provider_configs={"anthropic": {"api_key": "…"}})

async def main():
    async for chunk in client.chat.completions.acreate(
        model="anthropic:claude-3-5-sonnet-20240620",
        messages=[{"role": "user", "content": "What's the weather in Paris?"}],
        tools=[{
            "type": "function",
            "function": {
                "name": "get_weather",
                "description": "Fetch weather data",
                "parameters": {
                    "type": "object",
                    "properties": {
                        "city": {"type": "string"}
                    },
                    "required": ["city"]
                },
            }
        }],
        stream=True,
    ):
        if tool_calls := chunk.choices[0].delta.tool_calls:
            print("Tool call:", tool_calls[0].function.name, tool_calls[0].function.arguments)

asyncio.run(main())

AnthropicMessageConverter normalizes the native event stream so the caller never handles Anthropic-specific event types directly.

Async Streaming via the Base Bridge (Mistral)

If a provider lacks a native async stream, aisuite uses the base bridge. Here is how you stream from Mistral without blocking the event loop:

client = Client(provider_configs={"mistral": {"api_key": "…"}})

async for chunk in client.chat.completions.acreate(
    model="mistral:mistral-large-latest",
    messages=[{"role": "user", "content": "Write a haiku"}],
    stream=True,
):
    print(chunk.choices[0].delta.content, end="")

Because achat_completions_create_stream is inherited from the base class, the sync stream executes in a worker thread and yields chunks asynchronously.

Summary

• aisuite/provider.py defines the streaming contract and a generic async bridge using asyncio.Queue and run_in_executor.
• OpenAI streaming in aisuite/providers/openai_provider.py is a direct pass-through because the SDK already returns the correct chunk shape.
• Anthropic streaming in aisuite/providers/anthropic_provider.py relies on AnthropicMessageConverter.convert_stream_event (lines 62–154) to map mixed SDK events into ChatCompletionChunk objects.
• Gemini streaming in aisuite/providers/gemini_provider.py uses _StreamState (lines 31–92) to fold native chunks and synthesize tool calls and usage metadata.
• All other providers inherit the default bridge, ensuring every backend supports stream=True with a non-blocking async interface.
• The client façade in aisuite/client.py routes stream=True to the right provider method, always returning uniform ChatCompletionChunk instances.

Frequently Asked Questions

What object does aisuite return during streaming?

aisuite returns instances of aisuite.framework.chat_completion_chunk.ChatCompletionChunk for every provider. This object mirrors the OpenAI streaming delta format, exposing fields such as choices[0].delta.content and choices[0].delta.tool_calls.

Does every provider in aisuite support streaming?

Yes. Native providers such as OpenAI, Anthropic, and Gemini implement their own chat_completions_create_stream generators. Providers without a native implementation inherit the base class async bridge, which runs a synchronous stream in a background worker thread so that stream=True never blocks the event loop.

How does aisuite convert Anthropic streaming events into OpenAI chunks?

In aisuite/providers/anthropic_provider.py, each event from the Anthropic SDK is passed to AnthropicMessageConverter.convert_stream_event. This method—located at lines 62–154—normalizes tool-use start signals, text deltas, and final metadata into the standard ChatCompletionChunk shape, preserving usage counts and finish reasons.

Can I receive tool calls while streaming through aisuite?

Yes. Both Anthropic and Gemini handle tool-call streaming. Anthropic's native events are mapped to ChatCompletionChunk deltas containing tool_calls, while Gemini's _StreamState machine in aisuite/providers/gemini_provider.py emits complete tool-call chunks as soon as the full call is received from the underlying stream.

{
  "mcpServers": {
    "instagit": {
      "command": "npx",
      "args": ["-y", "instagit@latest"]
    }
  }
}