# How aisuite Handles Tool-Call Streaming with Incremental Delta Updates

> Discover how aisuite streams tool call updates incrementally. It builds full objects from delta pieces for live UI and delayed execution.

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

---

**aisuite streams model output in tiny "delta" pieces and builds complete tool-call objects only after the provider finishes sending its stream, ensuring live UI updates while postponing execution until the full response is assembled.**

Understanding how aisuite implements **tool-call streaming with incremental delta updates** reveals a carefully layered architecture that balances responsiveness with correctness. The andrewyng/aisuite codebase achieves this through three distinct layers: provider streaming, a core streaming bridge, and the engine event loop. Each layer has specific responsibilities for handling partial text fragments and deferring tool-call construction until the stream completes.

## Provider Streaming Layer: Emitting Deltas and Final Turns

Individual providers in aisuite implement the streaming protocol by yielding `StreamChunk` objects containing either incremental text or a complete turn. This design keeps provider implementations consistent across different model backends.

### OpenAI Provider Implementation

In [`platform/coworker/providers/openai_provider.py`](https://github.com/andrewyng/aisuite/blob/main/platform/coworker/providers/openai_provider.py), the `OpenAIProvider.stream` method collects `content` deltas and accumulates tool-call fragments throughout the stream. For each piece of content received, it yields a `StreamChunk(text_delta=...)`; only when the stream ends does it emit a final `StreamChunk(turn=AssistantTurn(...))` containing the assembled text and tool calls.

```python

# Simplified excerpt from openai_provider.py stream method

def stream(self, ...):
    for chunk in client.chat.completions.create(stream=True, ...):
        delta = chunk.choices[0].delta
        
        if delta.content:
            # Emit incremental text immediately

            yield StreamChunk(text_delta=delta.content)
        
        # Accumulate tool-call fragments internally

        # (not emitted until stream completion)

        # ...

    
    # Final chunk with complete AssistantTurn

    yield StreamChunk(
        turn=AssistantTurn(
            text=accumulated_text,
            tool_calls=assembled_tool_calls,
            finish_reason=finish_reason
        )
    )

```

### Anthropic Provider Implementation

The `AnthropicProvider.stream` method in [`platform/coworker/providers/anthropic_provider.py`](https://github.com/andrewyng/aisuite/blob/main/platform/coworker/providers/anthropic_provider.py) follows the same pattern, handling `text_delta` and `input_json_delta` events from Anthropic's streaming API. It similarly buffers partial tool-call payloads and produces a final `AssistantTurn` only after the stream concludes.

Both providers ensure that **partially-constructed tool calls never escape the provider layer**—the engine only sees complete, validated `ToolCall` objects.

## Core Streaming Bridge: Async Iterator Conversion

The `CoworkerEngine._astream` method in [`platform/coworker/engine.py`](https://github.com/andrewyng/aisuite/blob/main/platform/coworker/engine.py) transforms the blocking provider generator into an async iterator. This prevents the event loop from blocking during I/O operations while maintaining the ability to process deltas in real time.

The bridge launches the provider's `stream` in a separate thread, pushes `"chunk"` items onto an `asyncio.Queue`, and yields them to the async loop as they become available.

```python

# Conceptual flow from engine.py lines 55-87

async def _astream(self, provider, messages):
    queue = asyncio.Queue()
    
    def run_in_thread():
        for chunk in provider.stream(messages):
            queue.put_nowait(chunk)
        queue.put_nowait(None)  # Sentinel

    
    # Run blocking generator in thread pool

    asyncio.get_running_loop().run_in_executor(None, run_in_thread)
    
    # Yield chunks as async iterator

    while True:
        chunk = await queue.get()
        if chunk is None:
            break
        yield chunk

```

This pattern allows the engine to maintain responsiveness while consuming synchronous provider APIs.

## Engine Event Loop: Processing Deltas and Executing Tool Calls

The `CoworkerEngine._loop` method consumes `StreamChunk` objects and dispatches appropriate events. When a chunk contains `text_delta`, it immediately emits an `ASSISTANT_DELTA` event for live UI updates. When `chunk.turn` is present, it stores the completed turn, sends an `ASSISTANT_MESSAGE` event, and proceeds to `_handle_tool_calls`.

```python

# Logic from engine.py event loop (lines 6-12, 26-31)

async def _loop(self):
    async for chunk in self._astream(provider, messages):
        # Live delta: emit immediately for UI streaming

        if chunk.text_delta:
            self.emit(EventType.ASSISTANT_DELTA, {"text": chunk.text_delta})
        
        # Complete turn: store and process tool calls

        if chunk.turn:
            self.thread.add_turn(chunk.turn)
            self.emit(EventType.ASSISTANT_MESSAGE, {
                "text": chunk.turn.text,
                "tool_calls": chunk.turn.tool_calls
            })
            # Now safe to execute tools—fully parsed

            await self._handle_tool_calls(chunk.turn.tool_calls)

```

The critical insight: **tool-call execution only occurs after `chunk.turn` is received**, guaranteeing that all JSON parsing, validation, and assembly happens within the provider before any tool code runs.

## Data Structures: StreamChunk and AssistantTurn

The foundational types are defined in [`platform/coworker/providers/base.py`](https://github.com/andrewyng/aisuite/blob/main/platform/coworker/providers/base.py):

```python

# From base.py lines 51-57

@dataclass
class StreamChunk:
    text_delta: str | None = None      # Incremental content

    turn: AssistantTurn | None = None   # Complete turn (final chunk)

# From base.py lines 24-31

@dataclass
class AssistantTurn:
    text: str                          # Full assembled response

    tool_calls: list[ToolCall]         # Complete tool calls

    finish_reason: str | None

```

`ToolCall` represents a fully-specified function invocation with `name` and `arguments` fields, parsed from the provider's native format during stream accumulation.

## Practical Example: Consuming the Stream

Here's how to interact with aisuite's streaming output as an end user:

```python
from aisuite.platform.coworker.engine import CoworkerEngine
from aisuite.platform.coworker.providers.openai_provider import OpenAIProvider

engine = CoworkerEngine(
    provider=OpenAIProvider(api_key="..."),
    model="gpt-4",
    messages=[{
        "role": "user",
        "content": "Calculate 2^16 and format the result"
    }],
    registry=tool_registry,
    permissions=permission_manager,
)

async for event in engine.run():
    match event.type:
        case "assistant_delta":
            # Live typing effect—safe to render immediately

            print(event.data["text"], end="", flush=True)
        
        case "assistant_message":
            # Complete response with validated tool calls

            print(f"\n\nFull text: {event.data['text']}")
            for tc in event.data["tool_calls"]:
                print(f"🔧 Executing: {tc.name}({tc.arguments})")
        
        case "tool_result":
            print(f"📤 Result: {event.data['output']}")

```

The `assistant_delta` events provide sub-second responsiveness, while `assistant_message` delivers the definitive state including all tool calls ready for execution.

## Key Implementation Files

| File | Purpose |
|------|---------|
| [`platform/coworker/providers/base.py`](https://github.com/andrewyng/aisuite/blob/main/platform/coworker/providers/base.py) | `StreamChunk`, `AssistantTurn`, `ToolCall` dataclass definitions |
| [`platform/coworker/providers/openai_provider.py`](https://github.com/andrewyng/aisuite/blob/main/platform/coworker/providers/openai_provider.py) | OpenAI-specific streaming with delta accumulation |
| [`platform/coworker/providers/anthropic_provider.py`](https://github.com/andrewyng/aisuite/blob/main/platform/coworker/providers/anthropic_provider.py) | Anthropic-compatible streaming implementation |
| [`platform/coworker/engine.py`](https://github.com/andrewyng/aisuite/blob/main/platform/coworker/engine.py) | Async bridge, event loop, and tool-call dispatch |

## Summary

- **Providers yield `StreamChunk(text_delta=...)` for live updates** and `StreamChunk(turn=...)` only at stream completion
- **The async bridge** (`_astream`) converts blocking generators to non-blocking iterators without losing deltas
- **The engine loop** emits `ASSISTANT_DELTA` immediately but **defers tool execution** until the final `AssistantTurn` arrives
- **Tool-call integrity is guaranteed** because parsing and validation happen entirely within provider streaming code before any execution

## Frequently Asked Questions

### Why doesn't aisuite emit tool-call deltas like some other frameworks?

aisuite intentionally buffers tool-call fragments internally within each provider and only exposes complete `ToolCall` objects. This prevents partial or malformed JSON from reaching the execution layer, eliminating an entire class of runtime errors. The trade-off is slightly delayed tool-call visibility, but since tool calls require complete arguments to execute meaningfully, this design prioritizes correctness over premature exposure.

### How does aisuite handle extremely long streaming responses?

Because `StreamChunk` carries only incremental deltas during streaming, memory usage scales with the accumulated buffer size rather than the full response history. The async bridge's thread-to-queue pattern prevents blocking, and backpressure is naturally handled by the `asyncio.Queue` mechanism. For very long responses, the same accumulation pattern applies—text deltas emit live, while tool calls wait for completion.

### Can I access partial tool-call JSON if I need it for progressive UI rendering?

Not through the public `StreamChunk` API. The `AssistantTurn.tool_calls` field is only populated in the final chunk. If your use case requires showing "thinking" indicators during tool-call construction, you would need to extend the provider implementation to emit custom events during fragment accumulation—though this would bypass aisuite's safety guarantees about complete, valid tool calls.

### What happens if a stream is interrupted mid-way?

If the provider connection drops before emitting the final `StreamChunk(turn=...)`, the engine never receives tool calls to execute. Any accumulated text deltas were already emitted as `ASSISTANT_DELTA` events, but no `ASSISTANT_MESSAGE` or tool execution occurs. The application can detect this via timeout or connection error handling and optionally retry the request.