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

> Discover how aisuite's async support handles chat completions and tool execution efficiently across provider APIs, tool execution, and completion creation. Learn more.

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

---

**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`](https://github.com/andrewyng/aisuite/blob/main/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`](https://github.com/andrewyng/aisuite/blob/main/aisuite/provider.py) (lines 34-44) provides an awaitable interface even when the underlying SDK is synchronous:

```python

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

```python

# 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`](https://github.com/andrewyng/aisuite/blob/main/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:

```python

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

```python

# 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`](https://github.com/andrewyng/aisuite/blob/main/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:

```python

# 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

```python

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

```python
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`](https://github.com/andrewyng/aisuite/blob/main/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`](https://github.com/andrewyng/aisuite/blob/main/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`](https://github.com/andrewyng/aisuite/blob/main/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.