VoiceAgent

View as Markdown

VoiceAgent is an optional high-level runner in agentduet 1.0.0. It answers the call, opens a model session, streams PCM both ways, clears the buffer on barge-in, dispatches tool calls, and cleans up on hangup.

Integrations document the direct PCM bridge with agentduet and the provider SDK. Use this page when you want the same loop behind a thin adapter abstraction.

Install

$# try everything (no --pre; the extra already pins a pre-release adapters version)
$pip install "agentduet[adapters]"
$
$# production: one provider only (adapters package is still pre-release)
$pip install --pre "agentduet-adapters[gemini]"

Prefer the second for anything you deploy. [adapters] pulls all four providers’ SDKs; naming the provider keeps the install to what you use.

Credentials

VoiceAgent.from_env() reads AgentDuet connector credentials only:

$export AGENTDUET_API_KEY=your-connector-api-key
$export AGENTDUET_CONNECTOR_UUID=your-connector-uuid

Each model adapter reads its provider key separately (from the environment by default, or pass api_key= to the adapter constructor):

AdapterEnvironment variable
GeminiLiveGEMINI_API_KEY
GrokVoiceXAI_API_KEY
QwenVoiceDASHSCOPE_API_KEY
NovaSonicAWS credentials + AWS_REGION

Three-line Gemini agent

1from agentduet import VoiceAgent
2from agentduet_adapters.gemini import GeminiLive
3
4VoiceAgent.from_env().run(
5 GeminiLive(instruction="You are May, a warm phone concierge. Keep replies short.")
6)

Needs AGENTDUET_API_KEY, AGENTDUET_CONNECTOR_UUID, and GEMINI_API_KEY.

from_env() reads connector credentials and defaults sample_rate to 24000 (matching most realtime speech models). The SDK-wide CallAudioConfig default remains 16000 when you omit call_audio= on SessionManagerConfig.create().

Other adapters

1from agentduet_adapters.grok_voice import GrokVoice
2from agentduet_adapters.qwen import QwenVoice
3from agentduet_adapters.nova_sonic import NovaSonic
4
5VoiceAgent.from_env().run(GrokVoice())
6# VoiceAgent.from_env().run(QwenVoice())
7# VoiceAgent.from_env().run(NovaSonic())

Install the matching extra: agentduet-adapters[grok], [qwen], or [nova-sonic].

OpenAI Realtime is not packaged as an adapter yet. Use the OpenAI Realtime bridge tutorial.

Inbound and outbound

MethodRole
run(model)Blocking: connect and serve inbound calls until interrupted.
await serve(model)Async form of run().
call_out(dest, model, *, subscriber, ...)Blocking one-shot outbound call on its own client.
await dial(dest, model, *, subscriber, ...)Async form of call_out().
await place_call(dest, model, *, subscriber, ...)Outbound on the same client as a running serve(). Fire-and-forget.

Property: is_serving is True while serve() is running (required for place_call).

Trigger conditions on inbound

VoiceAgent(..., inbound=InboundCallMode.ALL) calls setup_trigger_conditions on startup with TriggerConditionsBuilder().inbound_call(...).build() (absolute replace). Message-flow toggles reset to their defaults. Pass inbound=None to leave the server-side configuration untouched. See Trigger Conditions and WhatsApp Messaging.

Tools, transcripts, usage

1from agentduet import VoiceAgent
2
3async def tools(name: str, args: dict) -> dict:
4 if name == "lookup_order":
5 return {"status": "shipped"}
6 return {"error": f"unknown tool {name}"}
7
8async def on_transcript(delta) -> None:
9 print(delta.role, delta.text)
10
11VoiceAgent.from_env(
12 tools=tools,
13 on_transcript=on_transcript,
14).run(GeminiLive(instruction="You help with orders. Keep replies short."))

Custom model adapter

Adapters are protocols. Any object with matching methods works; no base class required. Writing your own needs only agentduet (not agentduet-adapters):

1from agentduet import AudioOut, Interrupted, ToolCall, VoiceAgent
2
3class MyModel: # satisfies VoiceModel
4 async def open(self):
5 return MySession()
6
7class MySession: # satisfies ModelSession
8 async def push_audio(self, pcm: bytes): ...
9
10 def events(self):
11 # async iterator of AudioOut / Interrupted / ToolCall /
12 # TranscriptDelta / Usage
13 ...
14
15 async def send_tool_result(self, call_id, result): ...
16 async def close(self): ...
17
18VoiceAgent.from_env().run(MyModel())

Events VoiceAgent understands

EventFieldsReaction
AudioOutpcmSend to the call. Full send buffer drops the chunk.
Interrupted(none)clear_send_audio_buffer() (barge-in).
ToolCallid, name, argsDispatch tools= handler; return via send_tool_result.
TranscriptDeltatext, roleForward to on_transcript if set.
Usagetotal, input, output, detailForward to on_usage if set.

When to use a full bridge instead

Integrations use a direct PCM bridge with agentduet plus the provider SDK. Prefer that path when you need custom session lifecycle, OpenAI Agents / Google ADK shapes, resampling control, or multi-call topology. VoiceAgent is an optional abstraction over the same loop.