ivr_detection=True

ivr_detection=True

ivr_detection=True

from livekit.agents import Agent, function_tool, RunContext
from livekit import rtc

class IVRAgent(Agent):
    def __init__(self):
        super().__init__(
            instructions="You are an IVR assistant. Use DTMF detection to respond to user input."
        )
        # Ensure session is configured to detect IVRs when started
        self.session_options = {"ivr_detection": True}

    @function_tool()
    async def send_dtmf_code(self, context: RunContext, digit: str) -> str:
        """Tool to send a single DTMF digit from the agent (outbound)."""
        # Example: publish DTMF via room/local participant API
        # Replace with your actual RTC publish call
        await context.session.room.local_participant.publish_dtmf(code=digit)
        <http://context.session.info|context.session.info>(f"Sent DTMF: {digit}")
        return f"sent:{digit}"

    @function_tool()
    async def handle_dtmf_input(self, context: RunContext, digit: str, duration_ms: int) -> str:
        """Tool to handle an incoming DTMF digit detected in the call (inbound)."""
        <http://context.session.info|context.session.info>(f"Received DTMF '{digit}' duration {duration_ms}ms")

        # Map digits to actions / replies
        responses = {
            "1": "You selected sales. Transferring now.",
            "2": "You selected support. How can I help?",
            "3": "You selected billing. Please hold.",
            "*": "Returning to main menu."
        }
        reply = responses.get(digit, "Invalid option. Please try again.")
        # Generate a spoken reply using the agent/session TTS facility
        await context.session.generate_reply(instructions=reply)
        return reply

# Hook DTMF events from the room to call the tool
async def on_room_started(session):
    room = session.room

    @room.on("sip_dtmf_received")  # event name may vary; adapt to your SDK binding
    def _on_dtmf(event: rtc.SipDTMF):
        digit = event.digit
        duration = getattr(event, "duration_ms", 0)
        # Call the registered tool to process inbound DTMF
        # How you invoke a tool from an event depends on your agent runtime.
        session.invoke_tool("handle_dtmf_input", digit=digit, duration_ms=duration)

    # Optionally, to send a digit from code:
    # await room.local_participant.publish_dtmf(code="1")

send_dtmf_code

ivr_detection=True

SIP_OUTBOUND_TRUNK_ID

LIVEKIT_URL

LIVEKIT_API_KEY

LIVEKIT_API_SECRET

pip install livekit-api

import asyncio
import os
from livekit import api

async def make_call_and_select_butcher(phone_number: str, room_name: str = "grocery-call"):
    lkapi = api.LiveKitAPI(
        url=os.getenv("LIVEKIT_URL"),
        api_key=os.getenv("LIVEKIT_API_KEY"),
        api_secret=os.getenv("LIVEKIT_API_SECRET"),
    )

    outbound_trunk_id = os.getenv("SIP_OUTBOUND_TRUNK_ID")
    agent_name = "grocery-agent"

    try:
        # 1) Dispatch agent into the room (so an agent participant is present)
        await lkapi.agent_dispatch.create_dispatch(
            api.CreateAgentDispatchRequest(
                agent_name=agent_name,
                room=room_name
            )
        )
        print(f"Dispatched agent '{agent_name}' to room '{room_name}'")

        # 2) Place the outbound SIP call and wait until answered
        await lkapi.sip.create_sip_participant(
            api.CreateSIPParticipantRequest(
                room_name=room_name,
                sip_trunk_id=outbound_trunk_id,
                sip_call_to=phone_number,
                participant_identity="store-ivr",
                wait_until_answered=True
            )
        )
        print(f"Call placed to {phone_number} and answered (or waiting finished).")

        # 3) Wait briefly for IVR prompt to begin (adjust timing as needed)
        await asyncio.sleep(2.0)

        # 4) Send DTMF tone "2" from the agent/local participant to select butcher
        # Note: this call uses LiveKit API helper to publish DTMF for the room/participant.
        # Replace source_participant_identity if your participant identity differs.
        await lkapi.room.local_publish_dtmf(
            room_name=room_name,
            source_participant_identity=agent_name,  # identity of the agent/local participant
            code=2,
            digit="2"
        )
        print("Sent DTMF '2' to select butcher department.")

    except Exception as e:
        print("Error during call flow:", e)
    finally:
        await lkapi.aclose()

if __name__ == "__main__":
    phone = "+1234567890"  # grocery store phone number
    asyncio.run(make_call_and_select_butcher(phone))