# How to Stream Tool Calls with Incremental Deltas in aisuite

> Learn to stream tool calls with incremental deltas using aisuite. Seamlessly integrate LLM responses from Anthropic, OpenAI, and more into a unified protocol for real-time updates.

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

---

**aisuite normalizes LLM streaming responses—including tool calls—into a unified OpenAI-compatible chunk protocol using the `StreamChunk` data class, enabling real-time incremental deltas across Anthropic, OpenAI, and other providers.**

The `aisuite` library simplifies streaming tool calls with incremental deltas in aisuite applications by abstracting provider-specific SDKs. By converting raw streaming events into a standardized format, developers can consume partial text updates and tool-use fragments using a single interface regardless of the underlying model.

## Streaming Architecture and Core Components

The streaming implementation relies on three core primitives that normalize provider diversity into a consistent interface.

### The StreamChunk Data Class

Defined in [`platform/coworker/providers/base.py`](https://github.com/andrewyng/aisuite/blob/main/platform/coworker/providers/base.py), the **`StreamChunk`** class serves as the transport container for all incremental updates. It carries text deltas, optional `AssistantTurn` objects containing accumulated tool calls, and token usage statistics. This lightweight dataclass ensures every provider yields identical shapes regardless of native SDK differences.

### Provider-Specific Stream Adapters

Each LLM provider implements dedicated streaming methods that yield `StreamChunk` objects:

- **`OpenAIProvider`** in [`aisuite/providers/openai_provider.py`](https://github.com/andrewyng/aisuite/blob/main/aisuite/providers/openai_provider.py) implements `chat_completions_create_stream` and async variants, forwarding native `ChatCompletionChunk` objects through the unified converter.
- **`AnthropicProvider`** in both [`aisuite/providers/anthropic_provider.py`](https://github.com/andrewyng/aisuite/blob/main/aisuite/providers/anthropic_provider.py) and [`platform/coworker/providers/anthropic_provider.py`](https://github.com/andrewyng/aisuite/blob/main/platform/coworker/providers/anthropic_provider.py) implements `achat_anthropic_stream` and related methods, handling Anthropic's `MessageEvent` types including `content_block_delta` and `content_block_start`.

### Message Converters

**`AnthropicMessageConverter`** and **`OpenAICompliantMessageConverter`** translate native provider events into `StreamChunk` instances. Located within their respective provider files, these converters accumulate JSON fragments for tool calls (`input_json_delta`) until completion, then emit structured `ToolCall` objects.

## How Incremental Delta Streaming Works

The streaming flow follows a five-stage pipeline that delivers incremental updates to your application:

1. **Request dispatch**: The client in [`aisuite/client.py`](https://github.com/andrewyng/aisuite/blob/main/aisuite/client.py) routes the request to the configured provider when `stream=True` is passed to `client.chat.completions.create()`.

2. **Raw event streaming**: Providers receive native SDK iterators—OpenAI returns `ChatCompletionChunk` objects, while Anthropic yields `MessageEvent` types.

3. **Event normalization**: Message converters process each raw event, mapping `content_block_delta` text to `StreamChunk(text_delta=...)` and accumulating tool JSON fragments.

4. **Incremental yielding**: The provider generator yields each `StreamChunk` immediately upon processing, allowing callers to receive partial text ("Hel" → "lo") and emerging tool calls without waiting for the complete response.

5. **Final turn assembly**: When the stream concludes, the converter emits a final chunk containing the complete `AssistantTurn` with full tool call definitions, `finish_reason`, and usage statistics.

This architecture ensures tool calls appear in the stream as soon as the model generates them, enabling immediate execution while the LLM continues generating explanatory text.

## Basic Streaming with the High-Level Client

The simplest approach uses the unified client interface, which automatically handles provider selection and chunk normalization:

```python
from aisuite import client

# Define conversation and available tools

messages = [
    {"role": "user", "content": "What’s the weather in Paris?"},
]

tools = [
    {
        "type": "function",
        "function": {
            "name": "get_weather",
            "parameters": {
                "type": "object",
                "properties": {"location": {"type": "string"}}
            }
        }
    }
]

# Stream response with incremental deltas

for chunk in client.chat.completions.create(
    model="gpt-4o",
    messages=messages,
    tools=tools,
    stream=True,
):
    delta = chunk.choices[0].delta
    
    # Print text fragments as they arrive

    if delta.content:
        print(delta.content, end="", flush=True)
    
    # Handle tool calls immediately when emitted

    if delta.tool_calls:
        for tc in delta.tool_calls:
            print(f"\n🔧 Tool call: {tc.function.name} {tc.function.arguments}")

```

As implemented in [`tests/client/test_streaming.py`](https://github.com/andrewyng/aisuite/blob/main/tests/client/test_streaming.py), the `stream=True` parameter routes to the provider's streaming generator, yielding OpenAI-compatible chunks regardless of the backend provider.

## Async Streaming Patterns

For asynchronous applications, use `acreate` to receive an async iterator of chunks:

```python
import asyncio
from aisuite import client

async def stream_async():
    async for chunk in client.chat.completions.acreate(
        model="gpt-4o-mini",
        messages=[{"role": "user", "content": "Summarize this article"}],
        stream=True,
    ):
        delta = chunk.choices[0].delta
        if delta.content:
            print(delta.content, end="", flush=True)

asyncio.run(stream_async())

```

The async implementation mirrors the synchronous pattern while maintaining non-blocking I/O, as demonstrated in [`tests/client/test_async_client.py`](https://github.com/andrewyng/aisuite/blob/main/tests/client/test_async_client.py).

## Low-Level Provider Implementation

For direct provider access without the client abstraction, instantiate providers directly to inspect raw `StreamChunk` objects:

```python
from aisuite.providers.anthropic_provider import AnthropicProvider

provider = AnthropicProvider(api_key="YOUR_KEY")

# Access the native streaming generator

for chunk in provider.chat_completions_create_stream(
    "claude-3-5-sonnet-20240620",
    [{"role": "user", "content": "Translate ‘hello’ to French"}]
):
    if chunk.text_delta:
        print(chunk.text_delta, end="")
    
    if chunk.turn and chunk.turn.tool_calls:
        for tc in chunk.turn.tool_calls:
            print(f"\nTool call: {tc.name} {tc.arguments}")

```

The Anthropic provider's conversion logic in [`platform/coworker/providers/anthropic_provider.py`](https://github.com/andrewyng/aisuite/blob/main/platform/coworker/providers/anthropic_provider.py) (lines 68-86) handles the transformation of `content_block_delta` events into incremental text and tool JSON deltas.

## Inspecting Final Turn Metadata

Consume the complete stream to access final metadata including finish reasons and token usage:

```python
chunks = list(client.chat.completions.create(
    model="gpt-4o",
    messages=[{"role": "user", "content": "Execute analysis"}],
    stream=True,
))

final_chunk = chunks[-1].choices[0]
print(f"Finish reason: {final_chunk.finish_reason}")  # e.g., "tool_calls"

print(f"Prompt tokens: {final_chunk.usage.prompt_tokens}")

```

As verified in [`tests/providers/test_anthropic_streaming.py`](https://github.com/andrewyng/aisuite/blob/main/tests/providers/test_anthropic_streaming.py), the final chunk always contains the mapped `finish_reason` and usage statistics, regardless of whether the stream contained tool calls or text only.

## Summary

- **aisuite** implements streaming tool calls with incremental deltas by unifying provider SDKs into an OpenAI-compatible chunk protocol.
- The **`StreamChunk`** class in [`platform/coworker/providers/base.py`](https://github.com/andrewyng/aisuite/blob/main/platform/coworker/providers/base.py) serves as the standard transport container for text deltas, tool calls, and usage data.
- **Provider adapters** in [`aisuite/providers/openai_provider.py`](https://github.com/andrewyng/aisuite/blob/main/aisuite/providers/openai_provider.py) and [`aisuite/providers/anthropic_provider.py`](https://github.com/andrewyng/aisuite/blob/main/aisuite/providers/anthropic_provider.py) convert native events into incremental chunks.
- Tool calls appear in the stream as soon as JSON fragments are accumulated, enabling immediate execution alongside partial text generation.
- Both synchronous and asynchronous streaming patterns expose identical interfaces through `client.chat.completions.create()` and `acreate()`.

## Frequently Asked Questions

### What is the StreamChunk class in aisuite?

The **`StreamChunk`** class is a lightweight data container defined in [`platform/coworker/providers/base.py`](https://github.com/andrewyng/aisuite/blob/main/platform/coworker/providers/base.py) that standardizes streaming responses across all LLM providers. It carries incremental text updates via `text_delta`, optional `AssistantTurn` objects containing `ToolCall` lists, and token usage metadata, ensuring consistent access to streaming deltas regardless of the underlying API.

### How does aisuite handle tool call deltas from Anthropic?

According to the source in [`platform/coworker/providers/anthropic_provider.py`](https://github.com/andrewyng/aisuite/blob/main/platform/coworker/providers/anthropic_provider.py), the **`AnthropicMessageConverter`** processes Anthropic's `input_json_delta` events by accumulating JSON fragments until the `content_block_stop` event occurs. Once complete, it constructs a `ToolCall` object with the parsed function name and arguments, then yields a `StreamChunk` containing the tool call—enabling incremental delivery of tool use data as it streams from Claude.

### Can I stream tool calls asynchronously with aisuite?

Yes. The client supports asynchronous streaming through the `acreate` method, which returns an async iterator yielding the same `StreamChunk` objects as the synchronous implementation. As shown in [`tests/client/test_async_client.py`](https://github.com/andrewyng/aisuite/blob/main/tests/client/test_async_client.py), you can use `async for chunk in client.chat.completions.acreate(...)` to process incremental deltas without blocking the event loop.

### Where is the streaming logic implemented in the aisuite source code?

The streaming architecture spans several key files: [`platform/coworker/providers/base.py`](https://github.com/andrewyng/aisuite/blob/main/platform/coworker/providers/base.py) defines the `StreamChunk` protocol; [`aisuite/providers/openai_provider.py`](https://github.com/andrewyng/aisuite/blob/main/aisuite/providers/openai_provider.py) and [`aisuite/providers/anthropic_provider.py`](https://github.com/andrewyng/aisuite/blob/main/aisuite/providers/anthropic_provider.py) implement provider-specific streaming loops; [`aisuite/client.py`](https://github.com/andrewyng/aisuite/blob/main/aisuite/client.py) provides the public API façade; and [`tests/client/test_streaming.py`](https://github.com/andrewyng/aisuite/blob/main/tests/client/test_streaming.py) demonstrates the integration patterns. The Anthropic-specific conversion logic resides in [`platform/coworker/providers/anthropic_provider.py`](https://github.com/andrewyng/aisuite/blob/main/platform/coworker/providers/anthropic_provider.py) around lines 68-86.