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

> Learn to implement streaming tool calls with incremental delta processing in aisuite. Normalize LLM provider chunks into unified async generators for real-time processing.

- Repository: [Andrew Ng/aisuite](https://github.com/andrewyng/aisuite)
- Tags: how-to-guide
- Published: 2026-07-30

---

**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`](https://github.com/andrewyng/aisuite/blob/main/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.

```python

# 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`](https://github.com/andrewyng/aisuite/blob/main/aisuite/providers/openai_provider.py), the implementation yields native chunks directly since they already conform to the OpenAI shape:

```python

# 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`](https://github.com/andrewyng/aisuite/blob/main/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:

```python

# 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`](https://github.com/andrewyng/aisuite/blob/main/aisuite/framework/chat_completion_chunk.py). This model supports incremental delta processing through the `ChoiceDelta` class:

```python
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`](https://github.com/andrewyng/aisuite/blob/main/aisuite/client.py) exposes streaming through `chat_completions_create`. When `stream=True`, it delegates to the provider's async stream method:

```python

# 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:

```python
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:

```python
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:

```python
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`](https://github.com/andrewyng/aisuite/blob/main/aisuite/provider.py) automatically makes sync providers async-capable, while the client treats all implementations identically.

## Summary

- **Provider abstraction** in [`aisuite/provider.py`](https://github.com/andrewyng/aisuite/blob/main/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`](https://github.com/andrewyng/aisuite/blob/main/aisuite/providers/anthropic_provider.py)), which emit incremental `ChoiceDelta` objects containing partial tool-call arguments.
- **Unified consumption** through [`aisuite/client.py`](https://github.com/andrewyng/aisuite/blob/main/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.