How aisuite Handles Tool-Call Streaming with Incremental Delta Updates

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

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

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

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

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

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	StreamChunk, AssistantTurn, ToolCall dataclass definitions
platform/coworker/providers/openai_provider.py	OpenAI-specific streaming with delta accumulation
platform/coworker/providers/anthropic_provider.py	Anthropic-compatible streaming implementation
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.

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