> For clean Markdown of any page, append .md to the page URL. > For a complete documentation index, see https://docs.agentduet.com/llms.txt. > For AI client integration (Claude Code, Cursor, etc.), connect to the MCP server at https://docs.agentduet.com/_mcp/server. # Pipecat This guide shows how to integrate the AgentDuet SDK with [Pipecat](https://github.com/pipecat-ai/pipecat) pipelines using the [`pipecat-agentduet`](https://github.com/AgentDuet/agentduet-pipecat) transport package. ## What this integration provides A telephony transport (`AgentDuetTransport`) that moves live audio and call lifecycle events between an AgentDuet phone or WhatsApp call and any Pipecat pipeline. Unlike traditional telephony setups in Pipecat (which require provisioning public endpoints, running FastAPI webhook servers, setting up ngrok tunnels, and configuring media frame serializers), the AgentDuet SDK connects **outbound** to the AgentDuet platform. Calls are delivered down that persistent connection automatically. ### Key Features - **Zero Webhook Hosting:** No public URLs, ngrok, or reverse proxies required. - **Native PCM Audio:** Streams mono 16-bit PCM at the line rate (8 kHz, 16 kHz, or 24 kHz) with no manual audio transcoding. - **Unified Event Lifecycle:** Automatic call answering on pipeline start, cleanup on pipeline completion, and cancellation on remote hangup. - **Built-in Barge-in:** Fully compatible with Pipecat's VAD and turn processing for interruption handling. --- ## Prerequisites - Python **3.12+** - An API key and connector UUID from [agentduet.com](https://agentduet.com) - A [Deepgram API Key](https://console.deepgram.com/) (used for STT and TTS) - A [Google AI Studio API Key](https://aistudio.google.com/apikey) (used for Gemini LLM) ```bash pip install pipecat-agentduet "pipecat-ai[silero,deepgram,google]" python-dotenv ``` --- ## Step 1: Create a project folder ```bash mkdir agentduet-pipecat-bot && cd $_ python3.12 -m venv .venv && source .venv/bin/activate pip install pipecat-agentduet "pipecat-ai[silero,deepgram,google]" python-dotenv ``` --- ## Step 2: Configure `.env` Create a `.env` file in your project directory: ```bash cat > .env << 'EOF' AGENTDUET_API_KEY=your-agentduet-api-key AGENTDUET_CONNECTOR_UUID=your-connector-uuid DEEPGRAM_API_KEY=your-deepgram-api-key GOOGLE_API_KEY=your-google-api-key EOF ``` --- ## Step 3: Write `voice_bot.py` Create `voice_bot.py` next to `.env`: ```python import asyncio import os import uuid from dotenv import load_dotenv from agentduet import IncomingCallNotification, SessionManager, SessionManagerConfig from pipecat.audio.vad.silero import SileroVADAnalyzer from pipecat.frames.frames import LLMRunFrame from pipecat.pipeline.pipeline import Pipeline from pipecat.pipeline.worker import PipelineWorker from pipecat.processors.aggregators.llm_context import LLMContext from pipecat.processors.aggregators.llm_response_universal import LLMContextAggregatorPair from pipecat.processors.audio.vad_processor import VADProcessor from pipecat.services.deepgram.stt import DeepgramSTTService from pipecat.services.deepgram.tts import DeepgramTTSService from pipecat.services.google.llm import GoogleLLMService from pipecat.turns.user_turn_processor import UserTurnProcessor from pipecat.workers.runner import WorkerRunner from pipecat_agentduet import AgentDuetTransport load_dotenv() async def run_call(sm: SessionManager, noti: IncomingCallNotification): # Open an AgentDuet session for the incoming call session = await sm.open_session(uuid.uuid4().hex, noti.subscriber) call = await session.process_call(noti) # Wrap the call with the Pipecat transport transport = AgentDuetTransport(call) # Initialize conversational context context = LLMContext([ { "role": "system", "content": ( "You are a friendly voice assistant on a telephone call. " "Keep responses concise, natural, and conversational." ), } ]) aggregators = LLMContextAggregatorPair(context) # Build the Pipecat pipeline pipeline = Pipeline([ transport.input(), VADProcessor(vad_analyzer=SileroVADAnalyzer()), UserTurnProcessor(), DeepgramSTTService(api_key=os.environ["DEEPGRAM_API_KEY"]), aggregators.user(), GoogleLLMService( api_key=os.environ["GOOGLE_API_KEY"], settings=GoogleLLMService.Settings(model="gemini-3.5-flash-lite"), ), DeepgramTTSService(api_key=os.environ["DEEPGRAM_API_KEY"]), transport.output(), aggregators.assistant(), ]) worker = PipelineWorker(pipeline, idle_timeout_secs=None) # Greet caller immediately when call connects @transport.event_handler("on_dialin_connected") async def on_connected(t, payload): context.add_message({ "role": "developer", "content": "Start by briefly greeting the caller and asking how you can help.", }) await worker.queue_frames([LLMRunFrame()]) runner = WorkerRunner(handle_sigint=False) await runner.add_workers(worker) await runner.run() async def main(): config = SessionManagerConfig.create( api_key=os.environ["AGENTDUET_API_KEY"], connector_uuid=os.environ["AGENTDUET_CONNECTOR_UUID"], ) async with SessionManager(config) as sm: print("Bot is running and waiting for incoming calls...") @sm.on_incoming_call async def on_call(noti: IncomingCallNotification): asyncio.create_task(run_call(sm, noti)) await sm.run_forever() if __name__ == "__main__": asyncio.run(main()) ``` --- ## Step 4: Run the bot Start your voice bot: ```bash python voice_bot.py ``` Call your AgentDuet provisioned phone number or WhatsApp number. The bot answers automatically, greets you, and carries on a real-time voice conversation with interruption support. --- ## Event Lifecycle You can hook into call lifecycle events on the `AgentDuetTransport` instance using `@transport.event_handler(event_name)`: | Event | Fires When | Payload Type | |---|---|---| | `on_client_connected` | Call becomes active (inbound or outbound) | `CallEventPayload` | | `on_client_disconnected` | Call terminates (inbound or outbound) | `CallEventPayload` | | `on_dialin_connected` | Inbound `answer()` succeeds | `CallEventPayload` | | `on_dialin_stopped` | Remote caller hangs up | `CallEventPayload` | | `on_dialin_error` | Inbound call fails to connect | `CommandResult` | | `on_call_state_updated` | Every state change on the underlying `Call` | `CallState` | | `on_before_disconnect` | Just before disconnect handlers run (for final logging/flushing) | `CallEventPayload` | | `on_error` | Underlying SDK error occurs | Error object | --- ## Audio Architecture `AgentDuetTransport` reads the call's audio configuration directly from the AgentDuet `Call` object (mono 16-bit PCM at 8 kHz, 16 kHz, or 24 kHz) and provisions the pipeline input/output streams to match. Pipecat automatically manages intermediate resampling between services (such as TTS output rates and STT input rates), ensuring audio crosses between the telephony network and your pipeline cleanly. ## Related - [pipecat-agentduet on GitHub](https://github.com/AgentDuet/agentduet-pipecat) - [Pipecat Documentation](https://docs.pipecat.ai/) - [Gemini Live](/integrations/gemini-live) - [LangGraph](/integrations/lang-graph) - [Voice Cloning](/integrations/voice-cloning) ## Next Step Explore integration with ElevenLabs.