Every phone call has two fragile boundaries: the pickup and the hang-up. In real-time voice agents, these edges are where conversations lose words. A caller starts speaking the millisecond they hear the line connect, often well before your client application has finished handshaking with the upstream speech API. Minutes later, as the conversation wraps up, the agent begins saying goodbye, only for the client to assume the exchange is over and abruptly sever the socket mid-sentence.
To see how words get lost at the edges, look at this scripted call from the companion repository, pmckeetx/barge-in-voice-agent, connecting to PolyAI's Dialog-RSN-1 API. Here, an early client implementation handles a routine hotel booking inquiry, stumbling at both ends:
(caller says pickup.wav) (session ready: sent 0.00 s of held audio, now live) Caller: Yes, I'm calling about my booking for Friday. The surname's Okafor. (look_up_booking returned for Okafor) Agent: Let me pull up your booking, the system can take a moment to load. Agent: I have your booking here, Ms. Okafor. You're in room 214 for two nights, checking in on Friday. Is there anything else you need to know? (caller says thats-all.wav) Caller: Great, that's everything I needed. Thanks very much. (end_call: Caller confirmed they had everything they needed and said goodbye.) (hangs up) first line: lost 1 of 12 words at the start (session ready 0.26 s after pickup); goodbye: played 1.0 s of "You're welcome, Ms. Okafor. Have a wonderful stay with us. Goodbye.", then hung up ... Notice the damage on both sides of the exchange. The agent heard "Yes, I'm calling about..." because the caller's initial "Hi" landed during the quarter-second window before the session was ready, and the client discarded it. At the end of the call, the agent composed a polite farewell, but the client slammed the line down after a static one-second sleep while the server was still streaming audio chunks over the wire. There isn't even an Agent: transcript line for the goodbye because the call ended before the reply did.
Neither failure is the model's fault. The speech recognizer is working fine, and the language model knows what to say. The breakdown happens entirely in client-side lifecycle management: the client assumes the call hasn't started until the network is ready, and assumes the call is finished before playback concludes.
To fix this, we can equip our front-desk agent (the Harborview Inn from the repository) with two changes: a circular audio buffer that starts capturing frames the moment the sound card activates, flushing them the instant the session is configured; and an event-driven hang-up routine that tracks the server's response lifecycle and drains the playback buffer before tearing down the connection.
When benchmarked against the live API on 10 November 2026 across 10 runs each, the naive client clipped the caller's first word on every single call and severed every goodbye. The buffered, event-driven client captured every word at both ends.
Code: pmckeetx/barge-in-voice-agent (call.py and pickup.py)
What you'll build
You will build call.py, a voice agent client that listens from the exact instant the line opens and refuses to hang up until its farewell has finished playing through the speakers. Under the hood, it:
- Holds up to 5 seconds of caller audio in memory while waiting for the session handshake.
- Exposes an
end_calltool configured with a spoken goodbye argument. - Waits up to 8 seconds for that goodbye to fully stream and play.
- Cancels the hang-up and keeps the line open if the caller talks over the goodbye.
To verify these behaviors reliably, you'll also use pickup.py, a test harness that feeds scripted caller audio into the pipeline from the very first frame without requiring a live microphone. Here are the old client and the new one on the same scripted call:
How it works
To understand why calls clip, we have to look closely at the mechanics of the session lifecycle.
A call begins with an initialization dance across the network. The client opens a WebSocket, the server answers with session.created once accepted, the client replies with session.update specifying tools and audio configurations, and the server confirms with session.updated. Until that final confirmation arrives, the session runs on its defaults (which on 10 November 2026 meant raw 24 kHz PCM input, per the quickstart).
In an isolated connection probe, that handshake is fast: session.updated arrived a median 234 ms after the connection began (measured over 20 runs on api.studio from the US Midwest). In the 20 scripted calls further down, sessions were ready 0.23 to 2.83 seconds after pickup (median 0.66 s), for reasons I didn't pin down. A caller saying "Hi, yes..." gets those first words out inside that window, and routing calls over mobile networks or SIP trunks only widens the gap.
The end of the call presents the inverse problem. When the model decides the interaction is complete, it triggers end_call. Rather than sending plain text, the server speaks the tool's goodbye argument directly. The tool invocation and the outgoing audio stream arrive simultaneously: in 10 test calls, the first audio chunk arrived a median 46 ms before the client received the end_call tool event.
The server then streams the farewell audio in 200 ms chunks at a natural speaking cadence, keeping the client's local buffer between 100 ms and 500 ms ahead of real-time playback. The final response.done event only arrives after the last chunk is dispatched (a median 436 ms later). Because the client only ever holds about half a second of audio at a time, a naive asyncio.sleep(1.0) hangs up a second into a goodbye that runs about 4 seconds: by then, only 1.35 to 1.59 s of it had reached the client in the probe calls.
Reliable orchestration comes down to two explicit checkpoints: wait for session.updated before routing live frames upstream, and wait for response.done followed by speaker drainage before closing the socket.
Our implementation builds directly on FrontDesk in agent.py, a base class providing booking lookups, a core listen() event loop, logging utilities, and barge-in handling. Call in call.py wraps this foundation to reinforce both edges.
Prerequisites
To run the project, ensure you have Python 3.13 and uv installed. You will also need a Dialog-RSN-1 API key from the API Keys tab in studio.poly.ai.
The scripted test suite requires no hardware. If you plan to test call.py interactively, use headphones alongside your microphone to prevent local echo from feeding back into the agent.
Set up the project
Clone the companion repository and enter the project directory:
git clone https://github.com/pmckeetx/barge-in-voice-agent.git cd barge-in-voice-agent Initialize your environment variables from the template:
cp .env.example .env # Your Dialog-RSN-1 API key: sign in at https://studio.poly.ai and create one in the API Keys tab. POLY_API_KEY= # Optional: the full realtime endpoint. Defaults to wss://api.studio.poly.ai/v1/realtime. # DIALOG_REALTIME_URL= Add your API key to POLY_API_KEY. Leave DIALOG_REALTIME_URL commented out unless your key is for a different PolyAI host. Next, install dependencies and execute the offline mock suite:
uv sync uv run pytest ... tests/test_agent.py ...... [ 28%] tests/test_audio.py .... [ 47%] tests/test_call.py ......... [ 90%] tests/test_live.py ss [100%] ======================== 19 passed, 2 skipped in 12.77s ======================== The offline tests use a simulated event server. To run the two skipped integration tests against live infrastructure, run POLYAI_LIVE=1 uv run pytest -m live. Note that this executes two real calls lasting around 70 seconds and consumes concurrent session quota.
Step 1: Start listening at pickup
The primary mistake in naive voice clients is tying hardware capture to network readiness. The sound card is local and ready immediately; the remote session takes multiple network round trips to establish.
To prevent dropping words during that window, instantiate Call before opening the WebSocket, capturing all inbound audio into an in-memory ring buffer immediately:
FRAME_S = 0.02 # the sound card and the scripted caller both hand over 20 ms frames BUFFER_CAP_S = 5.0 # the most of the caller's audio we hold while the session gets ready HANG_UP_CAP_S = 8.0 # the longest we wait for a goodbye to finish before hanging up anyway ... class Call(FrontDesk): ... def __init__(self, ws, speaker: Speaker, echo: EchoCanceller | None = None): super().__init__(ws, speaker, echo) self.held: deque[bytes] = deque(maxlen=round(BUFFER_CAP_S / FRAME_S)) # past the cap, the oldest frame goes self.live = False # True once the held audio has gone out and frames can go straight to the server self.goodbye: dict | None = None # the end_call call, while its goodbye is being spoken self.goodbye_spoken = asyncio.Event() self.hanging_up: asyncio.Task | None = None self.hung_up = asyncio.Event() async def send_audio(self, pcm: bytes): if self.hung_up.is_set(): return if self.live: await super().send_audio(pcm) else: self.held.append(pcm) The held queue stores 20 ms frames with a maximum size of 250 entries (5 seconds total). If the connection stalls for 10 seconds, the oldest frames drop off the back, so the agent hears the caller's last five seconds, not the stalest. Inside send_audio(), frames accumulate in held until self.live flips to True. Once live, frames route straight to FrontDesk.send_audio(), which encodes them into input_audio_buffer.append events. Once the call is marked as hung up, any trailing mic input is dropped safely.
With the buffer ready, we invert our startup sequence: activate the local microphone stream first, and initiate the remote connection second. Here's call() in call.py:
async def call(speakers: bool): """A live call on your microphone. The microphone opens first, then the connection.""" speaker = Speaker() echo = EchoCanceller() if speakers else None if echo: speaker.on_played = echo.played desk = Call(None, speaker, echo) with SoundCard(speaker) as card: # pickup: from here on, everything the caller says is kept async def stream_mic(): while True: await desk.send_audio(await card.queue.get()) mic = asyncio.create_task(stream_mic()) print("Line open. Start talking whenever you like; say goodbye when you're done.") async with connect() as ws: desk.ws = ws await desk.start() await desk.listen() # returns when the hang-up closes the socket mic.cancel() Opening SoundCard inside a context manager begins capturing audio instantly. The background stream_mic task pumps frames into desk.send_audio() before connect() is even dispatched. When the connection establishes, desk.ws is assigned, start() negotiates the session, and listen() takes over until teardown.
Step 2: Send the held audio once the session is configured
Buffering is only half the battle; how and when you flush that buffer determines whether the agent understands the speech.
Warning: Any audio forwarded before
session.updatetakes effect is processed under the session's default audio format (24 kHz PCM). In testing on 10 November 2026, sending 8 kHz μ-law telephony audio prior tosession.updatedfailed silently in 5 out of 5 runs: no speech turn triggered, no transcript was generated, and no server error was raised.
Here is start() handling the configuration handshake and flushing the buffered speech:
async def start(self): await self.ws.recv() # session.created: the session is open, but still on its default settings await self.send({ "type": "session.update", "session": { "type": "realtime", "instructions": CALL_INSTRUCTIONS, "tools": TOOLS + [END_CALL], "audio": {"input": {"format": {"type": "audio/pcm", "rate": 24000}, "turn_detection": {"type": "server_vad"}}}, }, }) updated = json.loads(await self.ws.recv()) if updated["type"] != "session.updated": raise RuntimeError(updated.get("error", {}).get("message", f"unexpected {updated['type']}")) held = len(self.held) while self.held: # frames that arrive during the flush join the end of the queue, so order holds await super().send_audio(self.held.popleft()) self.live = True self.log("", f"(session ready: sent {held * FRAME_S:.2f} s of held audio, now live)") The method waits for session.created, dispatches session.update with instructions, schemas, and codecs, and explicitly waits for session.updated. If the server rejects the configuration, it yields an error rather than an updated event; checking for this explicitly prevents pumping audio into a misconfigured session.
Waiting for session.updated costs about 50 ms, and with the 24 kHz PCM this front desk sends, it saves no words: 24 kHz audio sent before session.update was heard in full in 5 of 5 runs, and so was μ-law sent between session.update and session.updated. The wait is there for a refused update, and for the day you switch to phone audio.
When releasing the buffer, frames flush sequentially as fast as the socket can transmit them. You might worry that dumping multiple seconds of speech into the server in one burst would break turn detection or voice activity detection (VAD). In practice, it didn't. In 30 runs on 10 November 2026, with the line sent at real time or with 1.5 s or 3 s of it in one burst, all 12 words were transcribed every time, each as a single turn. The server catches up, and audio_end_ms, where turn detection puts the end of the caller's speech, came out the same as for audio sent at real time.
One curious quirk emerged during testing: the caller's surname ("Okafor") was transcribed as "O'For" in 6 out of 10 scripted calls with held audio and in none of the 10 without it, while the probes spelled it right in all 30 bursts. I can't explain it yet. Every other word came through, but if names matter on your calls, test your own audio.
Notice also how frame ordering is preserved during the flush: new microphone frames entering send_audio() append to the tail of held while the loop runs. Because there is no await between draining the final frame and setting self.live = True, no incoming live frame can bypass the queue.
Step 3: Give the agent an end_call tool with a spoken goodbye
To end the call naturally, the agent must be able to say goodbye and trigger the termination signal within the same conversational turn. Here's the tool, from call.py:
END_CALL = { "type": "function", "name": "end_call", "description": "Hang up. Use it when the caller has everything they need and is saying goodbye.", "parameters": { "type": "object", "properties": { "goodbye": { "type": "string", "poly_spoken": True, # the server speaks this argument, in the same response as the call "description": "The last thing you say, as you hang up: thank the caller by name, wish them a good stay, and say goodbye.", }, "reason": {"type": "string", "description": "Why the call is ending, for the log."}, }, "required": ["goodbye", "reason"], }, } CALL_INSTRUCTIONS = INSTRUCTIONS + "\nWhen the caller has everything they need, end the call with end_call." The poly_spoken: True property on goodbye makes this work. It tells the server to say that argument aloud while the tool runs, in the same response as the call.
Attempting this via prompt engineering (such as instructing the model to "say goodbye, then invoke the tool") fails (probed on 23 October 2026): the model says the line, finishes its turn, and the tool call never comes. Making goodbye a required parameter is equally important; per the Dialog-RSN-1 documentation, the model often skips an optional poly_spoken argument.
In 10 probe calls on 10 November 2026, the model called end_call with a spoken goodbye every time, and the goodbye and the call came back in one response. In the scripted calls further down, the goodbyes ran 3.6 to 5.1 seconds. Because the session closes immediately afterward, the client does not need to send a function_call_output back to the model under normal conditions.
Step 4: Hang up when the goodbye has played
Replacing a brittle asyncio.sleep() requires listening for the exact server events that denote completion. In call.py, handle() monitors this progression:
async def handle(self, event: dict): match event["type"]: case "response.function_call_arguments.done" if event["name"] == "end_call": self.goodbye = event self.log("", f"(end_call: {json.loads(event['arguments']).get('reason', '')})") self.hanging_up = asyncio.create_task(self.hang_up_after_goodbye()) return # not a booking lookup: FrontDesk's tool runner never sees it case "response.done" if self.goodbye and event["response"]["id"] == self.goodbye["response_id"]: if event["response"]["status"] == "completed": self.goodbye_spoken.set() # the server has sent the whole goodbye else: await self.keep_line_open() await super().handle(event) When end_call arrives, handle() stores the event details and spawns the teardown task. It intercepts the event before the default tool runner attempts to process it as a database lookup. When the matching response.done event arrives with a completed status, it sets the goodbye_spoken event. Any non-completed status indicates an interruption, transitioning us into step 5.
The teardown task then coordinates the physical exit:
async def hang_up_after_goodbye(self): try: async with asyncio.timeout(HANG_UP_CAP_S): await self.goodbye_spoken.wait() # every word of the goodbye has arrived await self.speaker.drain() # and the speaker has played the last of it except TimeoutError: self.log("", f"(the goodbye didn't finish within {HANG_UP_CAP_S:.0f} s: hanging up anyway)") await self.hang_up() ... async def hang_up(self): self.hung_up.set() self.log("", "(hangs up)") await self.ws.close() This ensures the client waits for the entire goodbye to be received (goodbye_spoken.wait()), and then waits for the local output hardware to play the remaining audio buffer (self.speaker.drain()).
Wrapping the shutdown in an 8-second asyncio.timeout prevents deadlocks if a network packet drops or the server hangs. The Dialog-RSN-1 API bills connected time until the WebSocket closes, per the limits and pricing page, so the cap also bounds what a stuck call costs.
Contrast this approach with the naive implementation:
async def hang_up_after_goodbye(self): await asyncio.sleep(1.0) await self.hang_up() Closing the connection after an arbitrary 1-second pause truncates the goodbye while most of it is still on its way. In the scripted runs, the caller heard a median 1.0 s of each goodbye.
One alternative you might consider is sleeping briefly and then checking whether the speaker buffer is empty. When I replayed the timing of the 10 probe goodbyes, this played every one in full, because the server never let the local queue run dry. But it waits on the wrong thing: an empty speaker queue only indicates that the client has played whatever it has received so far. If a network hiccup introduces a 100 ms transmission gap, an empty-buffer check will mistakenly conclude the goodbye is over and hang up mid-word.
Step 5: Keep the line open if the caller talks over the goodbye
Real callers don't always respect the end of a script. A customer will often chime in with "Wait, one more question!" right as the agent begins saying goodbye.
Whenever a user speaks over model audio, the server issues a cancellation. In 5 out of 5 tests where the caller interrupted half a second into the farewell, the server marked response.done as cancelled about 0.8 s after the caller started talking.
Rather than terminating the call, Call catches this event and recovers:
async def keep_line_open(self): """The caller talked over the goodbye, so the server cancelled it. Stay on the line and tell the model.""" self.hanging_up.cancel() call, self.goodbye = self.goodbye, None self.log("", "(the caller talked over the goodbye: staying on the line)") await self.send({ "type": "conversation.item.create", "item": {"type": "function_call_output", "call_id": call["call_id"], "output": json.dumps( {"hung_up": False, "why": "The caller started talking over your goodbye, so the call is still open."})}, }) # No response.create: the caller's new turn gets its own reply. The method cancels the pending hang-up task, clears the goodbye state, and posts a function_call_output explaining to the model that the call remains open because the caller interrupted. We avoid dispatching response.create here because the caller's new turn is already being processed by server VAD and will trigger its own response.
Here is the resulting flow during an interruption, run via uv run pickup.py --talk-over:
... Caller: Great, that's everything I needed. Thanks very much. (end_call: Caller confirmed they had everything they needed and said goodbye.) (caller says one-more-thing.wav) (barge-in: stopped the reply after 1.3 s) (the caller talked over the goodbye: staying on the line) Agent: Thank you, Ms- [cut off] Caller: Oh wait, one more thing. Is breakfast included? Agent: I don't have information about that. Breakfast is served from 7 to 10 in the Atrium. Is there anything else you need? (caller says thats-all.wav) Caller: Great, that's everything I needed. Thanks very much. (end_call: Caller confirmed they had everything they needed and said goodbye.) Agent: Thank you, Ms. Okafor. We hope you have a wonderful stay. Goodbye. (hangs up) ... The [cut off] line tracks the actual audio heard by the caller, and it's also what the server keeps in conversation history, thanks to the front desk's barge-in handling. The agent answers the follow-up question and only closes the call once the customer says goodbye a second time.
Step 6: Test both edges on a server that answers late
Relying on unit tests against instantaneous local mocks can blind you to race conditions. To stress-test these timing edges, the repository includes a configurable mock server (FakeRealtime) equipped with ready_after (to simulate network delays during session setup) and pace (to stream replies the way api.studio did on 10 November 2026: 300 ms up front, then real time, with response.done 0.4 s after the last chunk). It can also pause a reply partway through. Excerpt from the reply loop in tests/fake_realtime.py:
if n == 0 and fc: # the goodbye's first audio arrives just ahead of the call itself await send({"type": "response.function_call_arguments.done", "response_id": rid, "item_id": fc["id"], "call_id": fc["call_id"], "name": "end_call", "arguments": fc["arguments"]}) if self.stall and (n + 1) * 100 == self.stall[0]: await asyncio.sleep(self.stall[1]) # the stream pauses, as on a slow network if not self.pace: await asyncio.sleep(0.005) elif n >= LEAD_MS // 100 - 1: await Source：https://dev.to/pmckeetx/start-and-end-a-voice-agent-call-in-python-without-clipping-a-word-330iRelated posts
Building a Real-Time GPS Proximity Radar & Mobile Chat PWA with $0/Month Map API Costs
Show and tell: I open-sourced Ambient Memory, a local-first context layer for Cursor and Claude Desktop
AI Daily Digest November 13, 2026: Gemini 4 Hydrogen Goes Default, Google's Gemini Glasses Open Pre-orders, Codex Security Goes Open Source
Stop your voice agent going quiet after it refunds a caller
I Forked an Open-Source Mac Utility. Here's What I Changed, What I Kept, and When You Should Use the Original
