How aisuite's Async Support Works for Chat Completions and Tool Execution

aisuite implements async chat completions and tool execution through three coordinated layers: thread-offloaded provider APIs in Provider.achat_completions_create, coroutine-aware tool execution in Tools.aexecute_tool, and async orchestration in Completions.acreate.

The aisuite library provides a unified interface for multiple LLM providers, and its asynchronous architecture ensures that blocking I/O never stalls your event loop. This article explains exactly how the aisuite source code implements non-blocking chat completions and multi-turn tool execution.

Provider Async API: Thread-Offloaded Completion Methods

Every provider in aisuite inherits from the base Provider class in aisuite/provider.py. This class defines two key async methods that wrap synchronous SDK calls.

Single-Turn Async Completions

The achat_completions_create method in aisuite/provider.py (lines 34-44) provides an awaitable interface even when the underlying SDK is synchronous:

# aisuite/provider.py

async def achat_completions_create(self, model_id, messages, **kwargs):
    """Async wrapper for chat_completions_create using thread offloading."""
    return await asyncio.to_thread(
        self.chat_completions_create,
        model_id,
        messages,
        **kwargs
    )

The asyncio.to_thread call moves the blocking chat_completions_create into a worker thread, returning control to the event loop immediately.

Async Streaming Iterator

For streaming responses, achat_completions_create_stream (lines 58-89) converts any synchronous generator into an async iterator:

# aisuite/provider.py

async def achat_completions_create_stream(self, model_id, messages, **kwargs):
    """Convert synchronous streaming to async iterator via queue."""
    queue = asyncio.Queue()
    
    def sync_stream():
        for chunk in self.chat_completions_create_stream(model_id, messages, **kwargs):
            queue.put_nowait(chunk)
        queue.put_nowait(None)  # Sentinel

    
    # Run synchronous generator in thread

    asyncio.create_task(asyncio.to_thread(sync_stream))
    
    # Yield chunks as they arrive

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

Providers with native async support—such as OpenAI and Anthropic—can override these methods for true non-blocking I/O without thread overhead.

Tool Runner Async Wrapper: Coroutine-Aware Execution

The Tools class in aisuite/utils/tools.py handles tool execution that may involve both regular functions and async def coroutines.

Main Entry Point: aexecute_tool

The aexecute_tool method (lines 443-448) mirrors its synchronous counterpart:

# aisuite/utils/tools.py

async def aexecute_tool(self, tool_call, context=None):
    """Async version of execute_tool with full policy and tracing."""
    # ... context preparation ...

    result = await self._invoke_tool_async(tool_call, context)
    # ... event emission and formatting ...

    return result

Invocation Helper: _invoke_tool_async

The _invoke_tool_async method (lines 72-78) detects function type and dispatches accordingly:

# aisuite/utils/tools.py

async def _invoke_tool_async(self, tool_call, context):
    """Execute tool as coroutine or in thread based on inspection."""
    tool_func = self._resolve_tool(tool_call)
    
    if inspect.iscoroutinefunction(tool_func):
        # Native async function: await directly

        return await tool_func(**context.params)
    else:
        # Blocking function: offload to thread

        return await asyncio.to_thread(tool_func, **context.params)

This design ensures that async tools run without thread overhead, while sync tools remain compatible without blocking the event loop.

Client Async Orchestration: Multi-Turn Tool Loops

The Completions class in aisuite/client.py coordinates the complete async flow, including multi-turn conversations with tool calls.

Entry Point: acreate

The acreate method (lines 84-95) branches based on whether tool execution is required:

# aisuite/client.py

async def acreate(self, model, messages, tools=None, max_turns=None, **kwargs):
    """Async entry point for chat completions."""
    provider = self._resolve_provider(model)
    
    if max_turns and tools:
        # Multi-turn async tool loop

        return await self._atool_runner(
            provider, model, messages, tools, max_turns, **kwargs
        )
    
    # Single-turn async completion

    return await provider.achat_completions_create(
        model.split(":")[1], messages, **kwargs
    )

Async Tool Loop: _atool_runner

The _atool_runner method (lines 95-115) implements the complete conversation loop:

• Awaits provider.achat_completions_create for each LLM call
• Awaits tools_instance.aexecute_tool for each tool execution
• Continues until max_turns is reached or no more tool calls are requested

# aisuite/client.py (conceptual structure)

async def _atool_runner(self, provider, model, messages, tools, max_turns, **kwargs):
    tools_instance = Tools(tools)
    
    for turn in range(max_turns):
        response = await provider.achat_completions_create(
            model_id, messages, tools=tools_instance.schema, **kwargs
        )
        
        if not response.tool_calls:
            break
            
        messages.append(response.message)
        
        for tool_call in response.tool_calls:
            result = await tools_instance.aexecute_tool(tool_call)
            messages.append({
                "role": "tool",
                "tool_call_id": tool_call.id,
                "content": str(result)
            })
    
    return response

Complete Working Example

Here is a fully asynchronous implementation using aisuite's async support:

import asyncio
from aisuite import Client

async def main():
    client = Client({"openai": {"api_key": "sk-…"}})

    # Single async completion (no tool loop)

    resp = await client.chat.completions.acreate(
        model="openai:gpt-4o-mini",
        messages=[{"role": "user", "content": "What's the weather in Paris?"}],
    )
    print(resp.choices[0].message.content)

    # Multi-turn async tool execution

    def get_weather(city: str) -> str:
        """Return a fake weather report."""
        return f"The weather in {city} is sunny."

    async def fetch_forecast(city: str) -> dict:
        """Native async tool for API calls."""
        # Imagine real async HTTP call here

        return {"city": city, "temp": 22, "condition": "sunny"}

    resp = await client.chat.completions.acreate(
        model="openai:gpt-4o-mini",
        messages=[{"role": "user", "content": "Compare weather in Berlin and Munich."}],
        tools=[get_weather, fetch_forecast],
        max_turns=3,
    )
    print(resp.choices[0].message.content)

asyncio.run(main())

Key Design Decisions

Decision	Implementation	Benefit
Thread offloading via asyncio.to_thread	Provider.achat_completions_create wraps sync SDKs	All providers gain async interface without code changes
Queue-based streaming	asyncio.Queue bridges sync generators to async iterators	Unified streaming API regardless of provider capability
Runtime coroutine detection	inspect.iscoroutinefunction in Tools._invoke_tool_async	Optimal execution path for each tool type
Mirrored sync/async paths	acreate/create, aexecute_tool/execute_tool	Consistent mental model, easy migration

Summary

• Provider layer: Provider.achat_completions_create and achat_completions_create_stream in aisuite/provider.py use asyncio.to_thread and asyncio.Queue to make any sync SDK non-blocking
• Tool layer: Tools.aexecute_tool and _invoke_tool_async in aisuite/utils/tools.py detect coroutines with inspect.iscoroutinefunction and execute natively or in threads
• Orchestration layer: Completions.acreate and _atool_runner in aisuite/client.py coordinate multi-turn async conversations with await on both LLM calls and tool execution

Frequently Asked Questions

Does aisuite require providers to implement native async support?

No. The base Provider class automatically makes sync providers async through asyncio.to_thread. Native async providers can override achat_completions_create for better performance.

How does aisuite handle mixed sync and async tools in the same conversation?

The Tools._invoke_tool_async method inspects each tool at runtime. async def functions are awaited directly; regular functions are executed via asyncio.to_thread. This happens transparently for each tool call.

What is the performance impact of thread offloading?

Thread offloading adds minimal overhead for I/O-bound operations like LLM API calls. For CPU-bound tools, consider implementing them as async def with asyncio primitives, or use asyncio.to_thread for blocking CPU work to avoid stalling the event loop.

Can I use async streaming with any provider?

Yes. The base Provider.achat_completions_create_stream implementation works with any provider that has a synchronous streaming method. Providers with native async streaming can override this for zero-thread streaming.

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