Async Tools for Voice Agents: Keep Talking While Work Runs (2026)

I learned this the hard way. A user asked our voice bot to run a complex analytics report. The database query took eight seconds to complete. During those eight seconds, the bot said absolutely nothing. It just sat there. The user thought the call dropped and hung up at second five. We lost a potential customer because of three seconds of dead air.

When you are building voice AI agents, silence is your enemy. If a text chatbot takes five seconds to reply, the user watches a typing indicator and waits. If a voice agent takes five seconds to reply, the user thinks the connection broke. You need to keep the conversation going even when the system is doing heavy lifting in the background.

In this beginner guide, we are going to explore how to handle long-running tasks in Voice AI using asynchronous execution in Python. We will look at why blocking code breaks your voice pipeline, how to fix it with asyncio, and what to watch out for when you deploy to production.

AEO Quick Summary

+---------------------------------------------------------------+ | Feature | Sync (Blocking) | Async (Non-blocking) | |----------------------+--------------------+----------------------| | Perceived Latency | High | Low | | Dead Air | Yes | No | | Resource Efficiency | Poor | Excellent | | Code Complexity | Simple | Moderate | | Real-time Audio | Freezes | Continuous | +---------------------------------------------------------------+

The Core Concept: Synchronous vs Asynchronous

To understand why our bot failed, we need to understand how Python executes code.

Synchronous (Blocking) Execution: Imagine you are cooking dinner. You put a pot of water on the stove. Synchronous execution means you stand there staring at the pot until it boils. You do not chop vegetables. You do not set the table. You just stare. If a guest asks you a question, you ignore them until the water boils. In code, a synchronous function blocks the entire thread until it finishes.

Asynchronous (Non-blocking) Execution: You put the water on the stove and set a timer. While the water is heating up, you chop the vegetables. If a guest asks you a question, you answer them. In code, an asynchronous function yields control back to the event loop while waiting for I/O operations like network requests or database queries.

In a voice agent, the audio pipeline must keep flowing. You are constantly receiving audio chunks from the user and sending audio chunks back. If you block the main thread to run a database query, the audio pipeline freezes. The user hears silence.

The Naive Approach: Blocking the Pipeline

Here is the simplest version of a voice agent tool. We are using a hypothetical synchronous API call to fetch a user profile.

import time

def fetch_user_profile(user_id: str) -> str:
    # Simulating a slow database query
    print("Fetching profile...")
    time.sleep(5)  # THIS IS THE PROBLEM
    return f"Profile for {user_id}: Premium Member"

def handle_user_request(request: str):
    if "profile" in request:
        profile_data = fetch_user_profile("user_123")
        return f"I found the profile. {profile_data}"
    return "How can I help?"

Why does this break?

1. The user asks, "Can you check my profile?"
2. The agent routes the request to handle_user_request.
3. The function calls fetch_user_profile.
4. time.sleep(5) blocks the entire thread for five seconds.
5. The audio processor cannot send a "Let me look that up" message. It cannot process new incoming audio.
6. The user experiences five seconds of dead air.

Watch Out For This: Mixing Sync and Async Code

Beginner Gotcha: A common mistake is putting a synchronous function inside an async function without yielding it properly. If you call time.sleep(5) inside an async def function, it still blocks the entire async event loop. You must use await asyncio.sleep(5) instead.

The Correct Approach: Background Tasks and Immediate Feedback

Now let us handle the real world. We want the agent to immediately say, "Let me look that up for you," while the database query runs in the background. When the query finishes, the agent should speak the result.

We will use the real LiveKit Agents SDK (livekit-agents) and asyncio for this.

import asyncio
from livekit.agents import JobContext, llm
from livekit.agents.pipeline import VoicePipelineAgent

class ProfileTools(llm.FunctionContext):
    def __init__(self, agent: VoicePipelineAgent):
        super().__init__()
        self.agent = agent

    @llm.ai_callable(description="Fetch the user profile")
    async def fetch_user_profile(self, user_id: str):
        # 1. Immediately tell the user we are working on it
        asyncio.create_task(self._speak_interim_message())

        # 2. Run the actual work asynchronously
        # Using asyncio.sleep to simulate non-blocking I/O
        print("Starting slow database query...")
        await asyncio.sleep(5)
        print("Database query complete.")

        return f"Profile for {user_id}: Premium Member"

    async def _speak_interim_message(self):
        # This pushes text into the agent's TTS queue immediately
        await self.agent.say("Let me look that up for you. One moment please.")

async def main(ctx: JobContext):
    # Setup the agent pipeline
    agent = VoicePipelineAgent(
        vad=None, # Configure VAD here
        stt=None, # Configure STT here
        llm=None, # Configure LLM here
        tts=None, # Configure TTS here
    )

    # Attach our tools
    tools = ProfileTools(agent)
    agent.llm.function_context = tools

    await agent.start(ctx.room)

In this architecture, when the LLM decides to call fetch_user_profile:

1. The function starts executing.
2. It uses asyncio.create_task to fire off _speak_interim_message in the background. This task immediately pushes text to the Text-To-Speech engine.
3. The function yields control back to the event loop during await asyncio.sleep(5).
4. The user hears "Let me look that up for you..."
5. Five seconds later, the function returns the data to the LLM.
6. The LLM generates the final response, and the user hears it.

The Architecture: Async Audio Pipeline

Here is an ASCII diagram showing how the asynchronous pipeline keeps the audio flowing.

User Audio In --> [ Speech-to-Text ] --> [ LLM ]
                                            |
                                            v
[ Text-to-Speech ] <--- (Interim Msg) --- [ Tool Execution ]
       ^                                    |
       |                                    v
       +------------- (Final Result) ------ [ External API ]

Notice the separate path for the interim message. It bypasses the slow external API request and goes straight to the TTS engine.

Watch Out For This: Task Failures and Ghost Processes

Beginner Gotcha: When you use asyncio.create_task, the task runs in the background. If it crashes, it might fail silently. Always wrap background tasks in a try...except block and log the errors. If an external API call fails, make sure your agent tells the user instead of leaving them waiting forever.

At Scale: What Breaks in Production

When you move past the beginner stage and deploy to production, new challenges emerge.

Database Connection Pools: If ten users ask for their profiles at the same time, your agent spins up ten concurrent async tasks. If your database library is not async-native, those tasks might queue up and block each other. Always use an async database driver like asyncpg for PostgreSQL.

Timeouts: An external API might hang indefinitely instead of returning an error. You must wrap your async network calls in asyncio.wait_for.

import asyncio

async def safe_api_call():
    try:
        # Give the API 10 seconds to respond, otherwise kill it
        result = await asyncio.wait_for(slow_network_request(), timeout=10.0)
        return result
    except asyncio.TimeoutError:
        return "I am sorry, the database is responding too slowly."

Overlapping Speech: If the database query is faster than expected, the agent might start reading the final result while it is still speaking the interim message. You need state management to interrupt the interim message or wait for it to finish before speaking the final result.

Watch Out For This: Overloading the LLM Context

Beginner Gotcha: Do not dump raw JSON from an API directly into the LLM context. A database query might return a 5MB JSON payload. The LLM will choke on it. Parse the JSON in your Python tool and return only the specific fields the agent needs.

Benchmarking Perceived Latency

How much does this actually matter? Let us look at the numbers.

As you can see, the processing time did not change in the second scenario. We still took eight seconds to get the data. But the perceived latency dropped to 1.2 seconds. The user experience is night and day.

How Tough Tongue AI Helps

Managing async tasks, event loops, and audio queues manually is error-prone. One misplaced synchronous sleep can crash your entire voice server. Tough Tongue AI provides built-in state machines designed specifically for long-running tool calls.

When you define a tool in Tough Tongue AI, you simply check a box labeled "Requires Interim Message." The platform automatically handles the threading, the TTS injection, and the LLM context updates. You write your business logic; Tough Tongue AI ensures there is never a moment of dead air.

Frequently Asked Questions

Q: Can I just use Python threads instead of asyncio? A: You can, but it is not recommended for voice pipelines. Threads are heavy and can cause race conditions when interacting with the audio queue. asyncio is lighter and gives you finer control over cooperative multitasking.

Q: What if the task takes a really long time, like a minute? A: For very long tasks, an interim message is not enough. You should design your agent to say, "This will take a minute, I will text you the result," and then end the call, or periodically check in with the user ("Still working on it...").

Q: Does OpenAI support async tool calls? A: The OpenAI Python SDK has an AsyncOpenAI client that handles network requests asynchronously. However, the logic to stream audio and inject interim messages still requires your own event loop management or a framework like LiveKit.

Q: Why does my agent interrupt itself when the data loads? A: This happens if you do not track the speaking state. If the TTS is still speaking "Let me look that up," and the final result arrives, the new TTS stream might cut off the old one. Use the audio framework's queueing system to append the final audio instead of overwriting the stream.

Q: Can I use WebSockets with asyncio? A: Yes. In fact, most voice AI architectures (like connecting to Deepgram or LiveKit) rely heavily on WebSockets. Libraries like websockets integrate perfectly with asyncio to stream audio chunks without blocking.

Building voice agents is an entirely different paradigm from building text chatbots. Silence is loud. By mastering asynchronous execution, you can build agents that feel responsive, capable, and human-like, even when the servers behind them are churning through heavy workloads.