Pipecat

import asyncio
import os
from dotenv import load_dotenv
from agentduet import (
    IncomingCallNotification,
    SessionManager,
    SessionManagerConfig,
    new_session_id,
)
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(
        noti.session_id or new_session_id(), noti.subscriber_id
    )
    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):
            await run_call(sm, noti)
        await sm.run_forever()
if __name__ == "__main__":
    asyncio.run(main())