How to Implement Streaming Tool Calls with Incremental Delta Processing in aisuite

aisuite normalizes streaming tool calls across LLM providers by converting native chunks into OpenAI-shaped deltas through provider-specific converters, enabling real-time processing of incremental text and tool-call arguments via unified async generators.

aisuite is a unified interface for multiple LLM providers that implements streaming tool calls with incremental delta processing to deliver real-time chat completions. This architecture allows developers to consume streaming responses containing both text fragments and partial tool-call arguments through a standardized async iterator, regardless of whether the underlying provider uses native async APIs or synchronous streaming endpoints.

Understanding the Streaming Architecture

Provider Base Class and Async Bridge

The foundation of streaming support resides in aisuite/provider.py, where the base Provider class establishes two critical pathways. For synchronous streaming implementations, providers override chat_completions_create_stream to return an iterator of raw SDK chunks. For asynchronous consumption, achat_completions_create_stream implements a threaded bridge that pushes sync chunks onto an asyncio.Queue, making any provider async-iterable without provider-specific async code.

# aisuite/provider.py – async bridge implementation

async def achat_completions_create_stream(self, model, messages, **kwargs):
    loop = asyncio.get_running_loop()
    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

This bridge guarantees that consumers can always use async for syntax even when the underlying SDK only supports synchronous streaming.

Provider-Specific Delta Normalization

Native-async providers like OpenAI and Anthropic override the bridge for true non-blocking I/O while implementing Δ (delta) normalization to convert proprietary chunk formats into a unified schema.

In aisuite/providers/openai_provider.py, the implementation yields native chunks directly since they already conform to the OpenAI shape:

# aisuite/providers/openai_provider.py

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

Anthropic requires more sophisticated handling in aisuite/providers/anthropic_provider.py. The AnthropicMessageConverter.convert_stream_event method maps Anthropic content-block indices to OpenAI tool-call indices, tracking state to emit incremental deltas:

# Excerpt from aisuite/providers/anthropic_provider.py

if event_type == "content_block_start":
    # First delta announces tool use, creating empty tool-call delta

    position = state.setdefault("tool_positions", {}) \
        .setdefault(event.index, len(state.get("tool_positions", {})))
    return self._delta_chunk(ChoiceDelta(
        tool_calls=[ChoiceDeltaToolCall(
            index=position,
            id=block.id,
            type="function",
            function=ChoiceDeltaFunction(name=block.name, arguments=""),
        )]
    ))

Subsequent content_block_delta events enrich the same tool-call with partial JSON arguments, producing incremental tool_calls deltas that surface as the stream progresses.

Unified Chunk Model

All providers yield ChatCompletionChunk objects defined in aisuite/framework/chat_completion_chunk.py. This model supports incremental delta processing through the ChoiceDelta class:

class ChatCompletionChunk:
    choices: List[StreamChoice]   # each choice holds a ChoiceDelta

class ChoiceDelta:
    role: Optional[str] = None
    content: Optional[str] = None
    tool_calls: Optional[List[ChoiceDeltaToolCall]] = None

ChoiceDelta can contain either text deltas (content) or tool-call deltas (tool_calls), mirroring the OpenAI streaming protocol and allowing downstream code to treat every provider uniformly.

Client Consumption

The public API in aisuite/client.py exposes streaming through chat_completions_create. When stream=True, it delegates to the provider's async stream method:

# aisuite/client.py – streaming entry point

async def chat_completions_create(self, *, model, messages, stream=False, **kwargs):
    provider = self._get_provider()
    if stream:
        async for chunk in provider.achat_completions_create_stream(model, messages, **kwargs):
            yield chunk
    else:
        return await provider.achat_completions_create(model, messages, **kwargs)

Implementing Streaming Tool Calls

Streaming with OpenAI (Native Async)

For OpenAI models, streaming tool calls with incremental delta processing works immediately since the provider yields native chunks without conversion:

import asyncio
from aisuite.client import AISuiteClient

async def stream_openai():
    client = AISuiteClient(provider_key="openai", config={"api_key": "sk-…"})
    async for chunk in client.chat_completions_create(
        model="gpt-4o-mini",
        messages=[
            {"role": "user", "content": "What's the weather in Paris?"},
        ],
        stream=True,
    ):
        # Print incremental text

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

        # Capture tool-call deltas as they arrive

        for tc in chunk.choices[0].delta.tool_calls or []:
            print(f"\nTool call: {tc.function.name}, args so far: {tc.function.arguments}")

asyncio.run(stream_openai())

Streaming with Anthropic (Incremental Delta Processing)

Anthropic's Claude models require the converter to normalize streaming tool calls. The first content_block_start event creates the tool-call delta with an empty arguments string, while subsequent content_block_delta events stream partial JSON:

import asyncio
from aisuite.client import AISuiteClient

async def stream_anthropic():
    client = AISuiteClient(provider_key="anthropic", config={"api_key": "…"})
    async for chunk in client.chat_completions_create(
        model="claude-3-5-sonnet-20240620",
        messages=[
            {"role": "user", "content": "Search for current AI headlines"},
        ],
        stream=True,
    ):
        # Text deltas

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

        # Incremental tool-call arguments

        for tc in chunk.choices[0].delta.tool_calls or []:
            print(f"\nPartial tool-call: {tc.function.name}")
            print(f"JSON fragment: {tc.function.arguments}")

asyncio.run(stream_anthropic())

You'll observe the arguments field growing incrementally until the final chunk marks the tool-call as complete.

Handling Synchronous-Only Providers

For providers that only implement synchronous streaming, the async bridge automatically handles thread management:

import asyncio
from aisuite.client import AISuiteClient

async def stream_sync_provider():
    # Provider only implements chat_completions_create_stream (sync)

    client = AISuiteClient(provider_key="some_sync_provider", config={})
    async for chunk in client.chat_completions_create(
        model="my-model",
        messages=[{"role": "user", "content": "Stream me something."}],
        stream=True,
    ):
        # Each chunk is already a normalized ChatCompletionChunk

        print(chunk.choices[0].delta.content or "")

The Provider.achat_completions_create_stream method spins a background thread automatically, eliminating the need for provider-specific async code.

Extending the Pattern for Custom Providers

To add a new provider supporting streaming tool calls with incremental delta processing:

1. Implement the streaming method – Override chat_completions_create_stream (sync) or achat_completions_create_stream (async) to yield raw provider chunks.

2. Create a message converter – Similar to AnthropicMessageConverter, implement conversion logic that:

   • Extracts finish reasons, usage statistics, and message fields
   • Builds incremental ChoiceDelta objects containing content or tool_calls
   • Maps provider-specific indices to OpenAI-compatible tool-call indices

3. Yield normalized chunks – Return ChatCompletionChunk objects using the _delta_chunk helper pattern to ensure compatibility with the unified client interface.

The default bridge in aisuite/provider.py automatically makes sync providers async-capable, while the client treats all implementations identically.

Summary

• Provider abstraction in aisuite/provider.py provides a threaded async bridge for any sync stream, while native-async providers override for optimal performance.
• Delta normalization occurs in provider-specific converters (e.g., AnthropicMessageConverter.convert_stream_event in aisuite/providers/anthropic_provider.py), which emit incremental ChoiceDelta objects containing partial tool-call arguments.
• Unified consumption through aisuite/client.py delivers a standard async generator of ChatCompletionChunk objects, enabling real-time UI updates and progressive tool-call handling across all supported LLM providers.
• OpenAI-shaped schema ensures that ChoiceDelta can carry either text or tool-call deltas, standardizing the streaming interface regardless of underlying provider protocols.

Frequently Asked Questions

What is the difference between text deltas and tool-call deltas in aisuite?

Text deltas contain incremental content fragments in the content field of ChoiceDelta, while tool-call deltas populate the tool_calls list with ChoiceDeltaToolCall objects containing partial arguments strings. Both types arrive through the same streaming interface, allowing your application to handle real-time text display alongside progressive tool-call argument accumulation.

How does aisuite handle providers that don't support native async streaming?

The base Provider class implements achat_completions_create_stream as an async bridge that runs the synchronous chat_completions_create_stream iterator in a background thread, feeding chunks through an asyncio.Queue. This ensures any provider becomes async-iterable without requiring provider-specific async implementations.

Can I implement incremental delta processing for a custom LLM provider?

Yes. Create a converter class that transforms your provider's native streaming events into ChatCompletionChunk objects containing ChoiceDelta instances. Track state to map your native indices to OpenAI-compatible tool-call indices, yielding incremental updates for both content and tool_calls fields as data arrives from the underlying API.

How do I accumulate partial tool-call arguments across multiple chunks?

Maintain a dictionary keyed by tool-call index (from ChoiceDeltaToolCall.index) in your application code. For each chunk, append the arguments string to the corresponding index entry. The final chunk for each tool-call will contain the complete JSON argument string, at which point you can parse it and execute the corresponding function.

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