deepgram-python-voice-agent

# Using Deepgram Voice Agent (Python SDK) Full-duplex voice agent runtime: STT + LLM (think) + TTS + function calling over a single WebSocket at `agent.deepgram.com/v1/agent/converse`. ## When to use this product - You want an **interactive voice assistant**: user speaks, agent thinks, agent speaks, interruptions allowed. - You want **function / tool calling** triggered by the conversation. - You want Deepgram to host the orchestration (vs wiring STT + LLM + TTS yourself). **Use a different skill when:** - One-way transcription → `deepgram-python-speech-to-text` or `deepgram-python-conversational-stt`. - One-way synthesis → `deepgram-python-text-to-speech`. - Analytics on finished audio → `deepgram-python-audio-intelligence`. - Managing reusable agent configs (persisted on the server) → `deepgram-python-management-api`. ## Authentication ```python from dotenv import load_dotenv load_dotenv() from deepgram import DeepgramClient client = DeepgramClient() ``` Header: `Authorization: Token <api_key>`. Base URL: `wss://agent.deepgram.com/v1/agent/converse`. ## Quick start ```python import threading, time from deepgram.core.events import EventType from deepgram.agent.v1.types import ( AgentV1Settings, AgentV1SettingsAgent, AgentV1SettingsAgentListen, AgentV1SettingsAgentListenProvider_V1, AgentV1SettingsAudio, AgentV1SettingsAudioInput, ) from deepgram.types.speak_settings_v1 import SpeakSettingsV1 from deepgram.types.speak_settings_v1provider import SpeakSettingsV1Provider_Deepgram from deepgram.types.think_settings_v1 import ThinkSettingsV1 from deepgram.types.think_settings_v1provider import ThinkSettingsV1Provider_OpenAi with client.agent.v1.connect() as agent: settings = AgentV1Settings( audio=AgentV1SettingsAudio( input=AgentV1SettingsAudioInput(encoding="linear16", sample_rate=24000), ), agent=AgentV1SettingsAgent( listen=AgentV1SettingsAgentListen( provider=AgentV1SettingsAgentListenProvider_V1(type="deepgram", model="nova-3"), ), think=ThinkSettingsV1( provider=ThinkSettingsV1Provider_OpenAi( type="open_ai", model="gpt-4o-mini", temperature=0.7, ), prompt="You are a helpful assistant. Keep replies brief.", ), speak=SpeakSettingsV1( provider=SpeakSettingsV1Provider_Deepgram(type="deepgram", model="aura-2-asteria-en"), ), ), ) agent.send_settings(settings) # MUST be first message after connect def on_message(m): if isinstance(m, bytes): # agent speech audio — play or append to output buffer return t = getattr(m, "type", "Unknown") if t == "ConversationText": print(f"[{getattr(m, 'role', '?')}] {getattr(m, 'content', '')}") elif t == "UserStartedSpeaking": print(">> user speaking") elif t == "AgentThinking": print(">> agent thinking") elif t == "AgentStartedSpeaking": print(">> agent speaking") elif t == "AgentAudioDone": print(">> agent done") elif t == "FunctionCallRequest": handle_tool_call(m) agent.on(EventType.OPEN, lambda _: print("open")) agent.on(EventType.MESSAGE, on_message) agent.on(EventType.CLOSE, lambda _: print("close")) agent.on(EventType.ERROR, lambda e: print(f"err: {e}")) def send_audio(): for chunk in mic_chunks(): agent.send_media(chunk) threading.Thread(target=send_audio, daemon=True).start() agent.start_listening() # blocks ``` ## Event types (server → client) - `Welcome` — connection acknowledged - `SettingsApplied` — your `Settings` accepted - `ConversationText` — text of a turn (with `role`: `user` or `assistant`) - `UserStartedSpeaking` — VAD detected user - `AgentThinking` — LLM is working - `FunctionCallRequest` — tool/function call initiated by the model - `AgentStartedSpeaking` — TTS starting - Binary frames — audio chunks - `AgentAudioDone` — TTS finished for this turn - `Warning`, `Error` ## Client messages - Initial `Settings` (send first) - `Media` (binary audio frames in declared encoding/sample_rate) - `KeepAlive` (on long sessions) - Prompt / think / speak update messages (change mid-session) - User / assistant text injection - Function call response (reply to `FunctionCallRequest`) - `ForceEndTurn` (end an active user turn; requires a Deepgram V2/Flux listen provider) ## Reusable agent configurations You can persist the **`agent` block** of a Settings message server-side and reuse it by `agent_id`. `client.voice_agent.configurations.create` stores a JSON string representing the `agent` object only (listen / think / speak providers + prompt) — NOT the full `AgentV1Settings` payload. Do not send top-level Settings fields like `audio` to that API; those still go in the live Settings message at connect time. The returned `agent_id` replaces the inline `agent` object in future Settings messages. Managed via `client.voice_agent.configurations.*` — see `deepgram-python-management-api`. ## Dynamic mid-session adjustment You can change agent behavior **without disconnecting** by sending control messages on the live socket. Each method is available on the agent connection object (`agent` in the quick-start) for both sync and async clients. ```python from deepgram.agent.v1.types import ( AgentV1UpdatePrompt, AgentV1UpdateSpeak, AgentV1UpdateSpeakSpeak, # type alias accepting SpeakSettingsV1 or list AgentV1UpdateThink, AgentV1UpdateThinkThink, # type alias accepting ThinkSettingsV1 or list AgentV1InjectAgentMessage, AgentV1InjectUserMessage, AgentV1KeepAlive, ) from deepgram.types.speak_settings_v1 import SpeakSettingsV1 from deepgram.types.speak_settings_v1provider import SpeakSettingsV1Provider_Deepgram from deepgram.types.think_settings_v1 import ThinkSettingsV1 from deepgram.types.think_settings_v1provider import ThinkSettingsV1Provider_OpenAi # 1. Swap the LLM system prompt mid-conversation (e.g. escalate to a different persona) agent.send_update_prompt( AgentV1UpdatePrompt(prompt="You are now in expert escalation mode. Be precise and concise.") ) # Server replies with a `PromptUpdated` event when the new prompt is in effect. # 2. Swap the TTS voice without reconnecting (e.g. switch language or persona) agent.send_update_speak( AgentV1UpdateSpeak( speak=SpeakSettingsV1( provider=SpeakSettingsV1Provider_Deepgram( type="deepgram", model="aura-2-luna-en", ), ), ) ) # Server replies with a `SpeakUpdated` event. # 3. Swap the LLM provider/model (e.g. cheaper model for follow-ups) agent.send_update_think( AgentV1UpdateThink( think=ThinkSettingsV1( provider=ThinkSettingsV1Provider_OpenAi( type="open_ai", model="gpt-4o-mini", temperature=0.3, ), prompt="You are a helpful assistant. Keep replies brief.", ), ) ) # Server replies with a `ThinkUpdated` event. # 4. Force the agent to say something specific (without waiting for user audio) agent.send_inject_agent_message( AgentV1InjectAgentMessage(message="Quick reminder: your call is being recorded.") ) # Useful for proactive prompts, status updates, or scripted segues. # 5. Inject a user message (e.g. text input from a chat sidebar alongside voice) agent.send_inject_user_message( AgentV1InjectUserMessage(content="Schedule a follow-up for next Tuesday at 2pm.") ) # Server may reply with `InjectionRefused` if the agent is mid-utterance — retry after `AgentAudioDone`. # 6. Idle-period keep-alive (no payload required; the SDK fills in the type literal) agent.send_keep_alive(AgentV1KeepAlive()) # Or simply: agent.send_keep_alive() — the message arg is optional. # 7. End an active user turn immediately (for example, on push-to-talk release). # Requires a Deepgram V2/Flux listen provider; V1 returns FORCE_END_TURN_UNSUPPORTED. agent.send_force_end_turn() ``` Async client equivalents are identical but `await`-prefixed: ```python await agent.send_update_prompt(AgentV1UpdatePrompt(prompt="...")) await agent.send_inject_agent_message(AgentV1InjectAgentMessage(message="...")) await agent.send_force_end_turn() ``` ## Stream lifecycle & recovery Continuous voice agents need explicit handling for idle periods, stream pauses, and reconnects. **Pause / idle (no audio for several seconds):** stop calling `send_media`, but emit a `KeepAlive` every ~5 seconds. Without it, the server closes the socket at ~10 seconds of idle. ```python import threading, time stop = threading.Event() def keepalive_loop(): while not stop.is_set(): if stop.wait(5): return try: agent.send_keep_alive() except Exception: return # socket closed; outer loop will reconnect threading.Thread(target=keepalive_loop, daemon=True).start() ``` **Resume after pause:** just call `send_media` again. No control message is required — the agent picks up VAD on the next chunk. **Reconnect after disconnect (preserve conversation context):** `Settings` cannot be re-sent on the same closed socket; open a new connection and resend the same `Settings`. To carry conversation history forward, include it in the new `Settings.agent.context.messages` so the LLM resumes with prior turns: ```python from deepgram.agent.v1.types import ( AgentV1SettingsAgentContext, AgentV1SettingsAgentContextMessagesItem, AgentV1SettingsAgentContextMessagesItemContent, AgentV1SettingsAgentContextMessagesItemContentRole, ) # Build the new Settings with the captured prior turns context = AgentV1SettingsAgentContext( messages=[ AgentV1SettingsAgentContextMessagesItem( content=AgentV1SettingsAgentContextMessagesItemContent( role=AgentV1SettingsAgentContextMessagesItemContentRole.USER, content="Hi, I'd like to schedule a meeting.", ), ), AgentV1SettingsAgentContextMessagesItem( content=AgentV1SettingsAgentContextMessagesItemContent( role=AgentV1SettingsAgentContextMessagesItemContentRole.ASSISTANT, content="Sure — what day works best?", ), ), ], ) new_settings = settings.model_copy(update={"agent": settings.agent.model_copy(update={"context": context})}) # Open a fresh connection and replay with client.agent.v1.connect() as agent2: agent2.send_settings(new_settings) # ... same handlers + audio loop as before ``` The server emits a `History` message on connect when the SDK has captured prior turns; in Python you receive this as an `AgentV1History` object (wire `type` literal: `"History"`). Persist these turns in your application so a reconnect can rebuild `context.messages`. **Detect disconnects:** the `EventType.CLOSE` handler fires before the `with` block exits. Catch it and trigger your reconnect logic from there. Check `EventType.ERROR` payloads for cause (network drop vs server-initiated close vs warning). ## API reference (layered) 1. **In-repo reference**: `reference.md` — "Agent V1 Connect", "Voice Agent Configurations". 2. **AsyncAPI (WSS)**: https://developers.deepgram.com/asyncapi.yaml 3. **Context7**: library ID `/llmstxt/developers_deepgram_llms_txt`. 4. **Product docs**: - https://developers.deepgram.com/reference/voice-agent/voice-agent - https://developers.deepgram.com/docs/voice-agent - https://developers.deepgram.com/docs/configure-voice-agent - https://developers.deepgram.com/docs/voice-agent-message-flow ## Gotchas 1. **Pick the right auth scheme for the credential type.** API keys use `Authorization: Token <api_key>`. Temporary / access tokens (created via `client.auth.v1.tokens.grant()` or an equivalent server) use `Authorization: Bearer <access_token>`. The custom `DeepgramClient` in this repo accepts an `access_token` parameter and installs a Bearer override for all HTTP + WebSocket calls — see `src/deepgram/client.py`. 2. **Base URL is `agent.deepgram.com`, not `api.deepgram.com`.** 3. **Send `Settings` IMMEDIATELY after connect** — no audio before settings are applied. 4. **Listen/speak encoding + sample_rate must match** both your input audio and your playback path. 5. **Keepalive on long idle sessions**, otherwise the server closes. 6. **Function call responses are synchronous to the turn** — reply promptly. 7. **Provider types are tagged unions** (`ThinkSettingsV1Provider_OpenAi`, `SpeakSettingsV1Provider_Deepgram`, ...). Pick the right union variant; don't pass raw dicts. 8. **`socket_client.py` is temporarily frozen** (see `.fernignore` → `src/deepgram/agent/v1/socket_client.py`) and currently carries `_sanitize_numeric_types` plus the `construct_type` / broad-catch fixes — needed for unknown WS message shapes. Expected to be unfrozen during a future Fern regen and re-compared. ## Example files in this repo - `examples/30-voice-agent.py` - `examples/32-voice-agent-force-end-turn.py` — Force an active turn to end with a Flux listen provider - `tests/manual/agent/v1/connect/main.py` — live connection test ## Central product skills For cross-language Deepgram product knowledge — the consolidated API reference, documentation finder, focused runnable recipes, third-party integration examples, and MCP setup — install the central skills: ```bash npx skills add deepgram/skills ``` This SDK ships language-idiomatic code skills; `deepgram/skills` ships cross-language product knowledge (see `api`, `docs`, `recipes`, `examples`, `starters`, `setup-mcp`).