How to Stream Tool Calls with Incremental Deltas in aisuite

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, 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 implements chat_completions_create_stream and async variants, forwarding native ChatCompletionChunk objects through the unified converter.
• AnthropicProvider in both aisuite/providers/anthropic_provider.py and 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 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:

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

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.

Low-Level Provider Implementation

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

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

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, 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 serves as the standard transport container for text deltas, tool calls, and usage data.
• Provider adapters in aisuite/providers/openai_provider.py and 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 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, 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, 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 defines the StreamChunk protocol; aisuite/providers/openai_provider.py and aisuite/providers/anthropic_provider.py implement provider-specific streaming loops; aisuite/client.py provides the public API façade; and tests/client/test_streaming.py demonstrates the integration patterns. The Anthropic-specific conversion logic resides in platform/coworker/providers/anthropic_provider.py around lines 68-86.

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